/
test_http.py
116 lines (91 loc) · 3.74 KB
/
test_http.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import os
import sys
import time
from threading import Thread
from unittest2 import TestCase
from wsgiref.simple_server import make_server
try:
import json
except ImportError:
import simplejson as json
from mesh.standard import *
from mesh.transport.http import *
from fixtures import *
server = HttpServer([primary_bundle, secondary_bundle])
def http(method, data=None, resource='example', mimetype=None, **params):
path = params.pop('path', None)
if not path:
bundle = params.pop('bundle', 'primary')
version = params.pop('version', (1, 0))
path = '/%s/%d.%d/%s' % (bundle, version[0], version[1], resource)
if mimetype is None:
if method == GET:
mimetype = URLENCODED
elif data:
mimetype = JSON
return server.dispatch(method, path, mimetype, {}, params, data)
class TestHttpServer(TestCase):
def setUp(self):
storage.reset()
def test_not_found(self):
for attempt in ('/primary/1.0/wrong', '/primary/10.0/example', '/wrong/1.0/example'):
response = http(GET, path=attempt)
self.assertEqual(response.status, NOT_FOUND)
def test_json_request(self):
data = {'required_field': 'text'}
response = http(POST, json.dumps(data))
self.assertEqual(response.status, OK)
content = json.loads(response.content[0])
self.assertIsInstance(content, dict)
self.assertIn('id', content)
self.assertIsInstance(content['id'], int)
def test_standard_requests(self):
response = http(POST, json.dumps({}))
self.assertEqual(response.status, INVALID)
response = http(POST, json.dumps({'required_field': 'text'}))
self.assertEqual(response.status, OK)
content = json.loads(response.content[0])
self.assertIsInstance(content, dict)
self.assertIn('id', content)
id = content['id']
self.assertIsInstance(id, int)
response = http(GET, resource='example/%d' % id)
self.assertEqual(response.status, OK)
self.assertEqual(json.loads(response.content[0]), {'id': id, 'required_field': 'text', 'default_field': 1})
response = http(POST, json.dumps({'default_field': 3}), resource='example/%d' % id)
self.assertEqual(response.status, OK)
self.assertEqual(json.loads(response.content[0]), {'id': id})
response = http(GET, 'exclude=[required_field]', resource='example/%d' % id)
self.assertEqual(response.status, OK)
self.assertEqual(json.loads(response.content[0]), {'id': id, 'default_field': 3})
response = http(DELETE, resource='example/%d' % id)
self.assertEqual(response.status, OK)
self.assertEqual(json.loads(response.content[0]), {'id': id})
response = http(GET, resource='example/%d' % id)
self.assertEqual(response.status, GONE)
class _TestHttpClient(TestCase):
@classmethod
def setUpClass(cls):
r, w = os.pipe()
pid = os.fork()
if pid:
cls.shutdown_pipe = w
time.sleep(2)
else:
server = make_server('localhost', 8888, wsgi_server)
thread = Thread(target=server.serve_forever)
thread.start()
os.read(r, 1)
server.shutdown()
thread.join()
os._exit(0)
@classmethod
def tearDownClass(cls):
os.write(cls.shutdown_pipe, '0')
def setUp(self):
storage.reset()
def test_standard_requests(self):
client = HttpClient('localhost:8888', primary_bundle.specify((1, 0)))
#response = client.execute('example', 'create', data={})
response = client.execute('example', 'create', data={'required_field': 'text'})
print response