morrowa@51: # CloseTestPong.py
morrowa@51: # Tests the closing negotiation facilities of the BLIP 1.1 protocol
morrowa@51: 
morrowa@51: from BLIP import Listener
morrowa@51: 
morrowa@51: import logging
morrowa@51: import asyncore
morrowa@51: import unittest
morrowa@51: 
morrowa@51: class CloseTestPong(unittest.TestCase):
morrowa@51:     
morrowa@51:     def shouldClose(self):
morrowa@51:         logging.info("Allowed to close.")
morrowa@51:         return True
morrowa@51:     
morrowa@51:     def handleConnection(self, conn):
morrowa@51:         logging.info("Accepted connection.")
morrowa@51:         conn.onCloseRequest = self.shouldClose
morrowa@51:     
morrowa@51:     def handleRequest(self, req):
morrowa@51:         resp = req.response
morrowa@51:         resp.body = "Pong"
morrowa@51:         resp.send()
morrowa@51:     
morrowa@51:     def testClose(self):
morrowa@51:         listen = Listener(1337)
morrowa@51:         listen.onConnected = self.handleConnection
morrowa@51:         listen.onRequest = self.handleRequest
morrowa@51:         
morrowa@51:         try:
morrowa@51:             asyncore.loop()
morrowa@51:         except KeyboardInterrupt:
morrowa@51:             pass
morrowa@51: 
morrowa@51: 
morrowa@51: if __name__ == '__main__':
morrowa@51:     logging.basicConfig(level=logging.DEBUG)
morrowa@51:     unittest.main()