Some HTML

from __future__ import generators

import time
from twisted.web2.test.test_server import BaseCase
from twisted.web2 import resource
from twisted.internet import reactor, interfaces
from twisted.python import log

if interfaces.IReactorThreads(reactor, None) is not None:
    from twisted.web2.wsgi import WSGIResource as WSGI
else:
    WSGI = None

class TestError(Exception):
    pass

class TestContainer(BaseCase):
    wait_timeout = 10.0

    def flushErrors(self, result, error):
        log.flushErrors(error)
        return result
    
    def test_getContainedResource(self):
        """Test that non-blocking WSGI applications render properly.
        """
        def application(environ, start_response):
            status = '200 OK'
            response_headers = [('Content-type','text/html')]
            writer = start_response(status, response_headers)
            writer('<html>')
            return ['<h1>Some HTML</h1>',
                    '</html>']

        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": None}, '<html><h1>Some HTML</h1></html>'))

    def test_getBlockingResource(self):
        """Test that blocking WSGI applications render properly.
        """
        def application(environ, start_response):
            """Simplest possible application object"""
            status = '200 OK'
            response_headers = [('Content-type','text/html')]
            writer = start_response(status, response_headers)
            writer('<h1>A little bit')
            time.sleep(1)
            writer(' of HTML</h1>')
            time.sleep(1)
            return ['<p>Hello!</p>']

        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": None}, '<h1>A little bit of HTML</h1><p>Hello!</p>'))

    def test_responseCode(self):
        """
        Test that WSGIResource handles strange response codes properly.
        """
        def application(environ, start_response):
            status = '314'
            response_headers = [('Content-type','text/html')]
            writer = start_response(status, response_headers)
            return []

        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (314, {"Content-Length": 0}, ''))

    def test_errorfulResource(self):
        def application(environ, start_response):
            raise TestError("This is an expected error")
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (500, {}, None)).addBoth(self.flushErrors, TestError)

    def test_errorfulResource2(self):
        def application(environ, start_response):
            write = start_response("200 OK", {})
            write("Foo")
            raise TestError("This is an expected error")
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": None}, "Foo"), failure=True
            ).addBoth(self.flushErrors, TestError)
    
    def test_errorfulIterator(self):
        def iterator():
            raise TestError("This is an expected error")
        
        def application(environ, start_response):
            start_response("200 OK", {})
            return iterator()
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (500, {}, None)).addBoth(self.flushErrors, TestError)

    def test_errorfulIterator2(self):
        def iterator():
            yield "Foo"
            yield "Bar"
            raise TestError("This is also expected")
        
        def application(environ, start_response):
            start_response("200 OK", {})
            return iterator()
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": None}, "FooBar"), failure=True
            ).addBoth(self.flushErrors, TestError)

    def test_didntCallStartResponse(self):
        def application(environ, start_response):
            return ["Foo"]
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (500, {}, None)).addBoth(self.flushErrors, RuntimeError)

    def test_calledStartResponseLate(self):
        def application(environ, start_response):
            start_response("200 OK", {})
            yield "Foo"
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": None}, "Foo"))

    def test_returnList(self):
        def application(environ, start_response):
            write = start_response("200 OK", {})
            return ["Foo", "Bar"]
        
        return self.assertResponse(
            (WSGI(application), 'http://host/'),
            (200, {"Content-Length": 6}, "FooBar"))

    def test_readAllInput(self):
        def application(environ, start_response):
            input = environ['wsgi.input']
            out = input.read(-1)
            start_response("200 OK", {})
            return [out]
        
        return self.assertResponse(
            (WSGI(application), 'http://host/', {}, None, None, '', "This is some content"),
            (200, {"Content-Length": 20}, "This is some content"))

    def test_readInputLines(self):
        def application(environ, start_response):
            input = environ['wsgi.input']
            out = 'X'.join(input.readlines())
            start_response("200 OK", {})
            return [out]
        
        return self.assertResponse(
            (WSGI(application), 'http://host/', {}, None, None, '', "These\nare\nlines"),
            (200, {"Content-Length": 18}, "These\nXare\nXlines\n"))

    def test_readInputMixed(self):
        def application(environ, start_response):
            input = environ['wsgi.input']
            out = [input.read(5)]
            out += ["X", input.readline()]
            out += ["X", input.read(1)]
            out += ["X", input.readline()]
            
            start_response("200 OK", {})
            return out
        
        return self.assertResponse(
            (WSGI(application), 'http://host/', {}, None, None, '', "Line blah blah\nOh Line\n"),
            (200, {"Content-Length": 26}, "Line Xblah blah\nXOXh Line\n"))

class TestWSGIEnvironment(BaseCase):
    """
    Test that the WSGI container does everything we expect it to do
    with the WSGI environment dictionary.
    """
    def envApp(self, *varnames):
        """
        Return a WSGI application that writes environment variables.
        """
        def _app(environ, start_response):
            status = '200'
            response_headers = [('Content-type','text/html')]
            writer = start_response(status, response_headers)
            return ['%s=%r;' % (k, environ.get(k, '')) for k in varnames]
        return _app

    def assertEnv(self, uri, env, version=None):
        keys = env.keys()
        keys.sort()
        envstring = ''.join(['%s=%r;' % (k, v) for k, v in env.items()])
        self.assertResponse(
            (WSGI(self.envApp(*keys)), uri, None, None, version),
            (200, {}, envstring))

    def test_wsgi_url_scheme(self):
        """wsgi.url_scheme"""
        self.assertEnv('https://host/', {'wsgi.url_scheme': 'https'})
        self.assertEnv('http://host/', {'wsgi.url_scheme': 'http'})

    def test_SERVER_PROTOCOL(self):
        """SERVER_PROTOCOL"""
        self.assertEnv('http://host/', {'SERVER_PROTOCOL': 'HTTP/1.1'})

    def test_SERVER_PORT(self):
        """SERVER_PORT"""
        self.assertEnv('http://host/', {'SERVER_PORT': '80'})
        self.assertEnv('http://host:523/', {'SERVER_PORT': '523'})
        self.assertEnv('https://host/', {'SERVER_PORT': '443'})
        self.assertEnv('https://host:523/', {'SERVER_PORT': '523'})
        self.assertEnv('/foo', {'SERVER_PORT': '80'}, version=(1,0))

if WSGI is None:
    for cls in (TestContainer, TestWSGIEnvironment):
        setattr(cls, 'skip', 'Required thread support is missing, skipping')