A beginning framework to unit test CherryPy handlers and tools

unittest.py

import cherrypy
from cStringIO import StringIO
import logging
import os
import pytest
import json
from mock import Mock


URL_PREFIX = "/api"


def simple_dispatch(app, url, verb=None):
    """
        This method just walks the application root. It will not look up any ids.
    """
    ret = app.root
    if url.startswith(URL_PREFIX):
        url = url[len(URL_PREFIX):]
    for seg in url.strip("/").split("/"):
        if seg:
            ret = getattr(ret, seg)
    if verb:
        ret = getattr(ret, verb.upper())
    return ret


def prep_headers(headers):
    if headers is None:
        headers = {}

    if not isinstance(headers, dict):
        headers = dict(headers)

    return headers


def prepare_http_request(
    context, url, body=None, headers=None, method="POST", protocol=None
):
    """
        Assumptions:
            url is a full path, not a schema://hostname/path?query url.
            body is a string or json object

        """
    req = context.http_request

    # Make sure the module attributes are what we expect them to be
    cherrypy.serving.request = req
    cherrypy.serving.response = context.http_response

    # factor out the URL_PREFIX mount root, because that's what the dispatcher will do.
    url = url.replace(URL_PREFIX, "")
    req.path_info = url
    if method:
        req.method = method
    headers = prep_headers(headers)

    if body:
        if not isinstance(body, (str, unicode)):
            req.json_doc = body
            body = json.dumps(body)
            headers["Content-Type"] = "application/json"
        req.body = cherrypy._cpreqbody.RequestBody(StringIO(body), req.headers)
        headers["Content-Length"] = str(len(body))

    req.headers.update(headers)

    if protocol:
        req.protocol = protocol


from warnings import warn

# TODO Rename to be more descriptive
def handle_request(context, method, headers=None, *args, **kw):
    warn("This method is obsolete before it was committed")
    if headers is not None:
        context.http_request.headers = headers
    ret = method(*args, **kw)
    return ret


def run_request(
    http_request, url=URL_PREFIX, query="", body="", headers=None, method="GET"
):

    req = http_request

    reqc = RequestContext.from_request(req)
    reqc.path_info = url
    reqc.query = query
    reqc.method = method

    if headers is None:
        headers = reqc.headers
    else:
        reqc.headers.update(headers)
        headers = reqc.headers

    if "Host" not in headers:
        headers["Host"] = "server.example.com"

    if body:
        if not isinstance(body, (str, unicode)):
            # Short cut to bypass json.loads in tools
            req.json_doc = reqc.raw_body = body
            reqc.body = json.dumps(body)
            headers["Content-Type"] = "application/json"
        headers["Content-Length"] = str(len(body))

    # TODO Turn off the json encoders on output
    # TODO Turn off performance and stats tools

    args = reqc.run_context()
    req.run(*args)
    return http_response


__all__ = [
    "URL_PREFIX",
]