Subversion Repositories HomeAutomation

Rev

Rev 743 | Go to most recent revision | Blame | Compare with Previous | Last modification | View Log | SVN | RSS feed

  1. import logging as log
  2. log.basicConfig(level=log.DEBUG,
  3.                 format='%(asctime)s %(levelname)s %(message)s',
  4.                 filename=__name__ + '.log',
  5.                 filemode='w')
  6. import unittest
  7.  
  8. from TCPServer import TCPServer
  9. from TCPClient import TCPClient
  10. import time
  11.  
  12. class CanPktTests(unittest.TestCase):
  13.    
  14.     client = None
  15.    
  16.     server = None
  17.     serverNotifier = None
  18.     serverRunning = False
  19.  
  20.     def setUp(self):
  21.         """Call before every test case."""
  22.         pass
  23.        
  24.     def tearDown(self):
  25.         """Call after every test case."""
  26.         if self.serverRunning:
  27.             self.stopServer()
  28.    
  29.     class ServerNotifier():
  30.        
  31.         parent = None
  32.         UNDEFINED = 0
  33.         TERMINATE = 1
  34.        
  35.         def __init__(self, parent):
  36.             self.parent = parent
  37.    
  38.         def notify(self, msg):
  39.             if msg == self.TERMINATE:
  40.                 log.debug('Server terminate notify')
  41.                 self.parent.serverRunning = False
  42.    
  43.     def startServer(self):
  44.         self.serverNotifier = self.ServerNotifier(self)
  45.         self.server = TCPServer(27000, self.serverNotifier)
  46.         assert self.server.start() == True
  47.         self.serverRunning = True
  48.        
  49.     def stopServer(self):
  50.         assert self.serverRunning == True
  51.         self.server.stop()
  52.         waitTime = 0
  53.         while self.serverRunning:
  54.             log.debug('Waiting 0.2s for server shutdown..')
  55.             time.sleep(0.2)
  56.             waitTime += 0.2
  57.             if waitTime > 4:
  58.                 log.debug('Giving up!')
  59.                 assert False
  60.                 break
  61.        
  62.     def startClient(self):
  63.         self.client = TCPClient('localhost', 27000)
  64.         return self.client.connect(1)
  65.  
  66.     def stopClient(self):
  67.         self.client.disconnect()
  68.         time.sleep(0.1)
  69.    
  70.     def testATcpServer(self):
  71.         self.startServer()
  72.         self.stopServer()
  73.  
  74.     def testBTcpClientNegative(self):
  75.         assert self.startClient() == False
  76.         self.stopClient()
  77.  
  78.     def testCTcpServerWithSingleClient(self):
  79.         self.startServer()
  80.         self.startClient()
  81.         self.stopClient()
  82.         self.stopServer()
  83.  
  84.     def testDTcpConnection(self):
  85.         log.debug('testTcpConnection')
  86.         self.startServer()
  87.         assert self.startClient() == True
  88.         log.debug('Server sends hej')
  89.         self.server.writeAll('hej')
  90.         time.sleep(0.5)
  91.         resp = str(self.client.read())
  92.         log.debug('Client gets:' + resp)
  93.         assert (resp == '\'hej\'')
  94.         self.stopClient()
  95.         self.stopServer()
  96.