166 lines
4.1 KiB
Python
166 lines
4.1 KiB
Python
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)
|