Ticket #1743: test_wsgi.py

#!/usr/bin/python
import os
from cStringIO import StringIO
from nevow import rend, wsgi, loaders, tags, testutil
from wsgiref.handlers import BaseHandler

from trace import Trace

class WSGITestHandler(BaseHandler):

    def _write(self, data):
        self.transport.write(data)

    def _flush(self):
        self.transport.flush()

    def get_stdin(self):
        return self.stdin

    def get_stderr(self):
        return self.stderr

    def add_cgi_vars(self):
        pass
    
    def handle_error(self):
        raise

class WSGITestHandlerWithQuery(WSGITestHandler):
    def __init__(self, method='GET', path='/', query='', proto='HTTP/1.1'):
        self.stdin = StringIO(' '.join((method,path,proto)))
        self.stderr = StringIO()
        self.transport = StringIO()
        self.additional_env = {
            'HTTP_ACCEPT': 'text/xml,text/html,text/plain',
            'HTTP_HOST': 'localhost:8080',
            'PATH_INFO': path,
            'QUERY_STRING': query,
            'REMOTE_ADDR': '127.0.0.1',
            'REMOTE_HOST': 'localhost',
            'REQUEST_METHOD': method,
            'SERVER_NAME': 'localhost',
            'SERVER_PORT': '8080',
            'SERVER_PROTOCOL': proto,
        }
        self.os_environ = dict(os.environ.items())
        self.os_environ.update(self.additional_env)


class NevowIndex(rend.Page):
    addSlash = True
    docFactory = loaders.stan(tags.html())
    
    def childFactory(self, context, name):
        return NevowIndex()

class WSGITestCase(testutil.TestCase):
    def setUp(self):
        self.app = wsgi.createWSGIApplication(NevowIndex())
    
    def testWSGIData(self):
        handler = WSGITestHandlerWithQuery()
        handler.run(self.app)
    
    def testWSGIHeader(self):
        handler = WSGITestHandlerWithQuery(path='/a')
        handler.run(self.app)