jens@13: #!/usr/bin/env python
jens@13: # encoding: utf-8
jens@13: """
jens@13: BLIPListenerTest.py
jens@13: 
jens@13: Created by Jens Alfke on 2008-06-04.
jens@13: This source file is test/example code, and is in the public domain.
jens@13: """
jens@13: 
jens@13: from BLIP import Listener
jens@13: 
jens@13: import asyncore
jens@13: import logging
jens@13: import unittest
jens@13: 
jens@13: 
jens@13: class BLIPListenerTest(unittest.TestCase):
jens@13:     
jens@13:     def testListener(self):
morrowa@56:         def handleConnection(conn):
morrowa@56:             logging.info("Got new connection: %r", conn)
morrowa@56:             conn.ListenerTestNumRequests = 0
morrowa@56:         
jens@13:         def handleRequest(request):
jens@13:             logging.info("Got request!: %r",request)
morrowa@56:             request.connection.ListenerTestNumRequests += 1
jens@13:             body = request.body
jens@13:             assert len(body)<32768
jens@13:             assert request.contentType == 'application/octet-stream'
jens@13:             assert int(request['Size']) == len(body)
jens@13:             assert request['User-Agent'] != None
jens@13:             for i in xrange(0,len(request.body)):
jens@13:                 assert ord(body[i]) == i%256
jens@13:             
jens@13:             response = request.response
jens@13:             response.body = request.body
jens@13:             response['Content-Type'] = request.contentType
jens@13:             response.send()
morrowa@56:             if request.connection.ListenerTestNumRequests >= 50:
morrowa@56:                 request.connection.close()
jens@13:         
jens@13:         listener = Listener(46353)
morrowa@56:         listener.onConnected = handleConnection
jens@13:         listener.onRequest = handleRequest
jens@13:         logging.info("Listener is waiting...")
jens@13:         
jens@13:         try:
jens@13:             asyncore.loop()
jens@13:         except KeyboardInterrupt:
jens@13:             logging.info("KeyboardInterrupt")
jens@13: 
jens@13: if __name__ == '__main__':
morrowa@56:     logging.basicConfig(level=logging.INFO)
jens@13:     unittest.main()