wsme/tests/test_cornice.py

import unittest
import json

import webtest

from cornice import Service
from cornice import resource
from pyramid.config import Configurator

from wsme.types import text, Base, HostRequest
from wsmeext.cornice import signature


class User(Base):
    id = int
    name = text

users = Service(name='users', path='/users')


@users.get()
@signature([User])
def users_get():
    return [User(id=1, name='first')]


@users.post()
@signature(User, body=User)
def users_create(data):
    data.id = 2
    return data


divide = Service(name='divide', path='/divide')


@divide.get()
@signature(int, int, int)
def do_divide(a, b):
    return a / b

needrequest = Service(name='needrequest', path='/needrequest')


@needrequest.get()
@signature(bool, HostRequest)
def needrequest_get(request):
    assert request.path == '/needrequest', request.path
    return True


class Author(Base):
    authorId = int
    name = text


@resource.resource(collection_path='/author', path='/author/{authorId}')
class AuthorResource(object):
    def __init__(self, request):
        self.request = request

    @signature(Author, int)
    def get(self, authorId):
        return Author(authorId=authorId, name="Author %s" % authorId)

    @signature(Author, int, body=Author)
    def post(self, authorId, data):
        data.authorId = authorId
        return data

    @signature([Author], text)
    def collection_get(self, where=None):
        return [
            Author(authorId=1, name="Author 1"),
            Author(authorId=2, name="Author 2"),
            Author(authorId=3, name="Author 3")
        ]


def make_app():
    config = Configurator()
    config.include("cornice")
    config.include("wsmeext.cornice")
    config.scan("test_cornice")
    return config.make_wsgi_app()


class WSMECorniceTestCase(unittest.TestCase):
    def setUp(self):
        self.app = webtest.TestApp(make_app())

    def test_get_json_list(self):
        resp = self.app.get('/users')
        self.assertEquals(
            resp.body,
            '[{"id": 1, "name": "first"}]'
        )

    def test_get_xml_list(self):
        resp = self.app.get('/users', headers={"Accept": "text/xml"})
        self.assertEquals(
            resp.body,
            '<result><item><id>1</id><name>first</name></item></result>'
        )

    def test_post_json_data(self):
        data = json.dumps({"name": "new"})
        resp = self.app.post(
            '/users', data,
            headers={"Content-Type": "application/json"}
        )
        self.assertEquals(
            resp.body,
            '{"id": 2, "name": "new"}'
        )

    def test_post_xml_data(self):
        data = '<data><name>new</name></data>'
        resp = self.app.post(
            '/users', data,
            headers={"Content-Type": "text/xml"}
        )
        self.assertEquals(
            resp.body,
            '<result><id>2</id><name>new</name></result>'
        )

    def test_pass_request(self):
        resp = self.app.get('/needrequest')
        assert resp.json is True

    def test_resource_collection_get(self):
        resp = self.app.get('/author')
        assert len(resp.json) == 3
        assert resp.json[0]['name'] == 'Author 1'
        assert resp.json[1]['name'] == 'Author 2'
        assert resp.json[2]['name'] == 'Author 3'

    def test_resource_get(self):
        resp = self.app.get('/author/5')
        assert resp.json['name'] == 'Author 5'

    def test_resource_post(self):
        resp = self.app.post(
            '/author/5',
            json.dumps({"name": "Author 5"}),
            headers={"Content-Type": "application/json"}
        )
        assert resp.json['authorId'] == 5
        assert resp.json['name'] == 'Author 5'

    def test_server_error(self):
        resp = self.app.get('/divide?a=1&b=0', expect_errors=True)
        self.assertEquals(resp.json['faultcode'], 'Server')
        self.assertEquals(resp.status_code, 500)

    def test_client_error(self):
        resp = self.app.get(
            '/divide?a=1&c=0',
            headers={'Accept': 'application/json'},
            expect_errors=True
        )
        print resp.body
        self.assertEquals(resp.json['faultcode'], 'Client')
        self.assertEquals(resp.status_code, 400)