#!/usr/bin/env python

 # encoding: utf-8

 """

 BLIPListenerTest.py

 Created by Jens Alfke on 2008-06-04.

 This source file is test/example code, and is in the public domain.

 """

 from BLIP import Listener

 import asyncore

 import logging

 import unittest

 class BLIPListenerTest(unittest.TestCase):

     def testListener(self):

         def handleConnection(conn):

             logging.info("Got new connection: %r", conn)

             conn.ListenerTestNumRequests = 0

         def handleRequest(request):

             logging.info("Got request!: %r",request)

             request.connection.ListenerTestNumRequests += 1

             body = request.body

             assert len(body)<32768

             assert request.contentType == 'application/octet-stream'

             assert int(request['Size']) == len(body)

             assert request['User-Agent'] != None

             for i in xrange(0,len(request.body)):

                 assert ord(body[i]) == i%256

             response = request.response

             response.body = request.body

             response['Content-Type'] = request.contentType

             response.send()

             if request.connection.ListenerTestNumRequests >= 50:

                 request.connection.close()

         listener = Listener(46353)

         listener.onConnected = handleConnection

         listener.onRequest = handleRequest

         logging.info("Listener is waiting...")

         try:

             asyncore.loop()

         except KeyboardInterrupt:

             logging.info("KeyboardInterrupt")

 if __name__ == '__main__':

     logging.basicConfig(level=logging.INFO)

     unittest.main()