root/branches/cherrypy-2.x/cherrypy/test/test_core.py

"""Basic tests for the CherryPy core: request handling."""

import test
test.prefer_parent_path()

import cherrypy
from cherrypy.lib import cptools, httptools
import types

import os
localDir = os.path.dirname(__file__)
log_file = os.path.join(localDir, "error.log")
log_access_file = os.path.join(localDir, "access.log")

defined_http_methods = ("OPTIONS", "GET", "HEAD", "POST", "PUT", "DELETE",
                        "TRACE", "CONNECT")


def setup_server():
    class Root:
        
        def index(self):
            return "hello"
        index.exposed = True
        
        def andnow(self):
            return "the larch"
        andnow.exposed = True
        
        def global_(self):
            pass
        global_.exposed = True
        
        def delglobal(self):
            del self.__class__.__dict__['global_']
        delglobal.exposed = True
        
        def defct(self, newct):
            newct = "text/%s" % newct
            cherrypy.config.update({'server.default_content_type': newct})
        defct.exposed = True
        
        def upload(self, file):
            return "Size: %s" % len(file.file.read())
        upload.exposed = True
    
    cherrypy.root = Root()


    class TestType(type):
        """Metaclass which automatically exposes all functions in each subclass,
        and adds an instance of the subclass as an attribute of cherrypy.root.
        """
        def __init__(cls, name, bases, dct):
            type.__init__(name, bases, dct)
            for value in dct.itervalues():
                if isinstance(value, types.FunctionType):
                    value.exposed = True
            setattr(cherrypy.root, name.lower(), cls())
    class Test(object):
        __metaclass__ = TestType


    class Params(Test):
        
        def index(self, thing):
            return repr(thing)
        
        def ismap(self, x, y):
            return "Coordinates: %s, %s" % (x, y)
        
        def default(self, *args, **kwargs):
            return "args: %s kwargs: %s" % (args, kwargs)


    class Status(Test):
        
        def index(self):
            return "normal"
        
        def blank(self):
            cherrypy.response.status = ""
        
        # According to RFC 2616, new status codes are OK as long as they
        # are between 100 and 599.
        
        # Here is an illegal code...
        def illegal(self):
            cherrypy.response.status = 781
            return "oops"
        
        # ...and here is an unknown but legal code.
        def unknown(self):
            cherrypy.response.status = "431 My custom error"
            return "funky"
        
        # Non-numeric code
        def bad(self):
            cherrypy.response.status = "error"
            return "bad news"


    class Redirect(Test):
        
        class Error:
            def _cp_on_error(self):
                raise cherrypy.HTTPRedirect("/errpage")
            
            def index(self):
                raise NameError()
            index.exposed = True
        error = Error()
        
        def index(self):
            return "child"
        
        def by_code(self, code):
            raise cherrypy.HTTPRedirect("somewhere else", code)
        
        def nomodify(self):
            raise cherrypy.HTTPRedirect("", 304)
        
        def proxy(self):
            raise cherrypy.HTTPRedirect("proxy", 305)
        
        def stringify(self):
            return str(cherrypy.HTTPRedirect("/"))


    class LoginFilter:
        
        def before_main(self):
            if cherrypy.config.get("auth.on", False):
                if not getattr(cherrypy.request, "login", None):
                    raise cherrypy.InternalRedirect("/internalredirect/login")

    class InternalRedirect(Test):
        
        _cp_filters = [LoginFilter()]
        
        def index(self):
            raise cherrypy.InternalRedirect("/")
        
        def relative(self):
            raise cherrypy.InternalRedirect("cousin")
        
        def cousin(self):
            return "I am a redirected page."
        
        def petshop(self, user_id):
            if user_id == "parrot":
                # Trade it for a slug when redirecting
                raise cherrypy.InternalRedirect('/image/getImagesByUser',
                                               "user_id=slug")
            elif user_id == "terrier":
                # Trade it for a fish when redirecting
                raise cherrypy.InternalRedirect('/image/getImagesByUser',
                                               {"user_id": "fish"})
            else:
                raise cherrypy.InternalRedirect('/image/getImagesByUser')
        
        def secure(self):
            return "Welcome!"
        
        def login(self):
            return "Please log in"


    class Image(Test):
        
        def getImagesByUser(self, user_id):
            return "0 images for %s" % user_id


    class Flatten(Test):
        
        def as_string(self):
            return "content"
        
        def as_list(self):
            return ["con", "tent"]
        
        def as_yield(self):
            yield "content"
        
        def as_dblyield(self):
            yield self.as_yield()
        
        def as_refyield(self):
            for chunk in self.as_yield():
                yield chunk


    class Error(Test):
        
        def custom(self):
            raise cherrypy.HTTPError(404, "No, <b>really</b>, not found!")
        
        def noexist(self):
            raise cherrypy.HTTPError(404, "No, <b>really</b>, not found!")
        
        def page_method(self):
            raise ValueError()
        
        def page_yield(self):
            yield "howdy"
            raise ValueError()
        
        def page_streamed(self):
            yield "word up"
            raise ValueError()
            yield "very oops"
        
        def cause_err_in_finalize(self):
            # Since status must start with an int, this should error.
            cherrypy.response.status = "ZOO OK"
        
        def log_unhandled(self):
            raise ValueError()
        
        def rethrow(self):
            """Test that an error raised here will be thrown out to the server."""
            raise ValueError()


    class Ranges(Test):
        
        def get_ranges(self, bytes):
            return repr(httptools.getRanges('bytes=%s' % bytes, 8))
        
        def slice_file(self):
            path = os.path.join(os.getcwd(), os.path.dirname(__file__))
            return cptools.serveFile(os.path.join(path, "static/index.html"))


    class Expect(Test):
        
        def expectation_failed(self):
            expect = cherrypy.request.headers.elements("Expect")
            if expect and expect[0].value != '100-continue':
                raise cherrypy.HTTPError(400)
            raise cherrypy.HTTPError(417, 'Expectation Failed')

    class Headers(Test):
        def default(self, headername): 
            """Spit back out the value for the requested header.""" 
            return cherrypy.request.headers[headername]
        
        def doubledheaders(self):
            # From http://www.cherrypy.org/ticket/165:
            # "header field names should not be case sensitive sayes the rfc.
            # if i set a headerfield in complete lowercase i end up with two
            # header fields, one in lowercase, the other in mixed-case."
            
            # Set the most common headers
            hMap = cherrypy.response.headers
            hMap['content-type'] = "text/html"
            hMap['content-length'] = 18
            hMap['server'] = 'CherryPy headertest'
            hMap['location'] = ('%s://127.0.0.1:%s/headers/'
                                % (cherrypy.request.remote_port,
                                   cherrypy.request.scheme))
            
            # Set a rare header for fun
            hMap['Expires'] = 'Thu, 01 Dec 2194 16:00:00 GMT'
            
            return "double header test"
        
        def ifmatch(self):
            val = cherrypy.request.headers['If-Match']
            cherrypy.response.headers['ETag'] = val
            return repr(val)
    
    
    class HeaderElements(Test):
        
        def get_elements(self, headername):
            e = cherrypy.request.headers.elements(headername)
            return "\n".join([str(x) for x in e])
    
    
    class Method(Test):
        
        def index(self):
            m = cherrypy.request.method
            if m in defined_http_methods:
                return m
            
            if m == "LINK":
                raise cherrypy.HTTPError(405)
            else:
                raise cherrypy.HTTPError(501)
        
        def parameterized(self, data):
            return data
        
        def request_body(self):
            # This should be a file object (temp file),
            # which CP will just pipe back out if we tell it to.
            return cherrypy.request.body

    class Divorce:
        """HTTP Method handlers shouldn't collide with normal method names.
        For example, a GET-handler shouldn't collide with a method named 'get'.
        
        If you build HTTP method dispatching into CherryPy, rewrite this class
        to use your new dispatch mechanism and make sure that:
            "GET /divorce HTTP/1.1" maps to divorce.index() and
            "GET /divorce/get?ID=13 HTTP/1.1" maps to divorce.get()
        """
        
        documents = {}
        
        def index(self):
            yield "<h1>Choose your document</h1>\n"
            yield "<ul>\n"
            for id, contents in self.documents.iteritems():
                yield ("    <li><a href='/divorce/get?ID=%s'>%s</a>: %s</li>\n"
                       % (id, id, contents))
            yield "</ul>"
        index.exposed = True
        
        def get(self, ID):
            return ("Divorce document %s: %s" %
                    (ID, self.documents.get(ID, "empty")))
        get.exposed = True

    cherrypy.root.divorce = Divorce()

    class Encoded:
        def default(self, *args):
            return "Encoded response"
        default.exposed = True

    cherrypy.root.encoded = Encoded()

    class Cookies(Test):
        
        def single(self, name):
            cookie = cherrypy.request.simple_cookie[name]
            cherrypy.response.simple_cookie[name] = cookie.value
        
        def multiple(self, names):
            for name in names:
                cookie = cherrypy.request.simple_cookie[name]
                cherrypy.response.simple_cookie[name] = cookie.value


    class ThreadLocal(Test):
        
        def index(self):
            existing = repr(getattr(cherrypy.request, "asdf", None))
            cherrypy.request.asdf = "rassfrassin"
            return existing
    
    cherrypy.config.update({
        'global': {
            'server.log_to_screen': False,
            'server.environment': 'production',
            'server.show_tracebacks': True,
            'server.max_request_body_size': 200,
            'server.max_request_header_size': 500,
        },
        '/flatten': {
            'server.log_file': log_file,
            'server.log_access_file': log_access_file,
        },
        '/params': {
            'server.log_file': log_file,
        },
        '/internalredirect/secure': {
            'auth.on': True,
        },
        '/error': {
            'server.log_file': log_file,
            'server.log_tracebacks': True,
        },
        '/error/page_streamed': {
            'stream_response': True,
        },
        '/error/cause_err_in_finalize': {
            'server.show_tracebacks': False,
        },
        '/error/custom': {
            'error_page.404': os.path.join(localDir, "static/index.html"),
        },
        '/error/noexist': {
            'error_page.404': "nonexistent.html",
        },
        '/error/log_unhandled': {
            'server.log_tracebacks': False,
            'server.log_unhandled_tracebacks': True,
        },
        '/error/rethrow': {
            'server.throw_errors': True,
        },
    })


#                             Client-side code                             #

import helper

class CoreRequestHandlingTest(helper.CPWebCase):
    
    def testParams(self):
        self.getPage("/params/?thing=a")
        self.assertBody("'a'")
        
        self.getPage("/params/?thing=a&thing=b&thing=c")
        self.assertBody("['a', 'b', 'c']")
        
        # Test friendly error message when given params are not accepted.
        ignore = helper.webtest.ignored_exceptions
        ignore.append(TypeError)
        try:
            self.getPage("/params/?notathing=meeting")
            self.assertInBody("index() got an unexpected keyword argument 'notathing'")
        finally:
            ignore.pop()
        
        # Test "% HEX HEX"-encoded URL, param keys, and values
        self.getPage("/params/%d4%20%e3/cheese?Gruy%E8re=Bulgn%e9ville")
        self.assertBody(r"args: ('\xd4 \xe3', 'cheese') "
                        r"kwargs: {'Gruy\xe8re': 'Bulgn\xe9ville'}")
        
        # Make sure that encoded = and & get parsed correctly
        self.getPage("/params/code?url=http%3A//cherrypy.org/index%3Fa%3D1%26b%3D2")
        self.assertBody(r"args: ('code',) "
                        r"kwargs: {'url': 'http://cherrypy.org/index?a=1&b=2'}")
        
        # Test coordinates sent by <img ismap>
        self.getPage("/params/ismap?223,114")
        self.assertBody("Coordinates: 223, 114")
    
    def testStatus(self):
        self.getPage("/status/")
        self.assertBody('normal')
        self.assertStatus(200)
        
        self.getPage("/status/blank")
        self.assertBody('')
        self.assertStatus(200)
        
        self.getPage("/status/illegal")
        self.assertStatus(500)
        msg = "Illegal response status from server (781 is out of range)."
        self.assertErrorPage(500, msg)
        
        self.getPage("/status/unknown")
        self.assertBody('funky')
        self.assertStatus(431)
        
        self.getPage("/status/bad")
        self.assertStatus(500)
        msg = "Illegal response status from server ('error' is non-numeric)."
        self.assertErrorPage(500, msg)
    
    def testLogging(self):
        open(log_file, "wb").write("")
        open(log_access_file, "wb").write("")
        
        self.getPage("/flatten/as_string")
        self.assertBody('content')
        self.assertStatus(200)
        
        self.getPage("/flatten/as_yield")
        self.assertBody('content')
        self.assertStatus(200)
        
        data = open(log_access_file, "rb").readlines()
        address = data[0].split()[0]
        if address not in ['127.0.0.1', '::ffff:127.0.0.1', '::1']:
            self.fail('access log did not contain remote IP:\n%s' % (data[0]))
          
        haslength = False
        for k, v in self.headers:
            if k.lower() == 'content-length':
                haslength = True
        if haslength:
            self.assert_(data[0].endswith('] "GET %s/flatten/as_string HTTP/1.1" 200 7 "" ""\n'
                                          % self.prefix()))
        else:
            self.assert_(data[0].endswith('] "GET %s/flatten/as_string HTTP/1.1" 200 - "" ""\n'
                                          % self.prefix()))
        
        address = data[1].split()[0]
        if address not in ['127.0.0.1', '::ffff:127.0.0.1', '::1']:
            self.fail('access log did not contain remote IP:\n%s' % (data[0]))

        haslength = False
        for k, v in self.headers:
            if k.lower() == 'content-length':
                haslength = True
        if haslength:
            self.assert_(data[1].endswith('] "GET %s/flatten/as_yield HTTP/1.1" 200 7 "" ""\n'
                                          % self.prefix()))
        else:
            self.assert_(data[1].endswith('] "GET %s/flatten/as_yield HTTP/1.1" 200 - "" ""\n'
                                          % self.prefix()))
        
        data = open(log_file, "rb").readlines()
        self.assertEqual(data, [])
        
        ignore = helper.webtest.ignored_exceptions
        ignore.append(ValueError)
        try:
            # Test that tracebacks get written to the error log.
            self.getPage("/error/page_method")
            self.assertInBody("raise ValueError()")
            data = open(log_file, "rb").readlines()
            self.assertEqual(data[0][-41:], ' INFO Traceback (most recent call last):\n')
            self.assertEqual(data[6], '    raise ValueError()\n')
            
            # Test that unhandled tracebacks get written to the error log
            # if log_tracebacks is False but log_unhandled_tracebacks is True.
            self.getPage("/error/log_unhandled")