Python/BLIPListenerTest.py
author Jens Alfke <jens@mooseyard.com>
Sun Jul 13 10:42:50 2008 -0700 (2008-07-13)
changeset 22 8b883753394a
child 51 de59ce19f42e
permissions -rw-r--r--
* Fixed: Responses still pending when a connection closed were not calling their onComplete targets.
* Fixed: BLIPTestClient target failed to build because it didn't link against zlib.
* If TCPListener.bonjourServiceName is changed while the listener is open, it now re-publishes the service with the new name.
* Added a TCPListener.bonjourService property.
* Added a BLIPMessage.representedObject property.
* Fixed a memory leak.
jens@13
     1
#!/usr/bin/env python
jens@13
     2
# encoding: utf-8
jens@13
     3
"""
jens@13
     4
BLIPListenerTest.py
jens@13
     5
jens@13
     6
Created by Jens Alfke on 2008-06-04.
jens@13
     7
This source file is test/example code, and is in the public domain.
jens@13
     8
"""
jens@13
     9
jens@13
    10
from BLIP import Listener
jens@13
    11
jens@13
    12
import asyncore
jens@13
    13
import logging
jens@13
    14
import unittest
jens@13
    15
jens@13
    16
jens@13
    17
class BLIPListenerTest(unittest.TestCase):
jens@13
    18
    
jens@13
    19
    def testListener(self):
jens@13
    20
        def handleRequest(request):
jens@13
    21
            logging.info("Got request!: %r",request)
jens@13
    22
            body = request.body
jens@13
    23
            assert len(body)<32768
jens@13
    24
            assert request.contentType == 'application/octet-stream'
jens@13
    25
            assert int(request['Size']) == len(body)
jens@13
    26
            assert request['User-Agent'] != None
jens@13
    27
            for i in xrange(0,len(request.body)):
jens@13
    28
                assert ord(body[i]) == i%256
jens@13
    29
            
jens@13
    30
            response = request.response
jens@13
    31
            response.body = request.body
jens@13
    32
            response['Content-Type'] = request.contentType
jens@13
    33
            response.send()
jens@13
    34
        
jens@13
    35
        listener = Listener(46353)
jens@13
    36
        listener.onRequest = handleRequest
jens@13
    37
        logging.info("Listener is waiting...")
jens@13
    38
        
jens@13
    39
        try:
jens@13
    40
            asyncore.loop()
jens@13
    41
        except KeyboardInterrupt:
jens@13
    42
            logging.info("KeyboardInterrupt")
jens@13
    43
jens@13
    44
if __name__ == '__main__':
jens@13
    45
    logging.basicConfig(level=logging.INFO)
jens@13
    46
    unittest.main()