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
|
#!/usr/bin/env python
# sugar-lint: disable
import json
import select
from __init__ import tests
from sugar_network import client as local
from sugar_network.toolkit.router import route, Router, Request, Response
from sugar_network.toolkit import coroutine, http
class HTTPTest(tests.Test):
def test_Subscribe(self):
class CommandsProcessor(object):
events = []
@route('GET', cmd='subscribe')
def subscribe(self, request, response):
while CommandsProcessor.events:
coroutine.sleep(.1)
yield CommandsProcessor.events.pop(0) + '\n'
self.server = coroutine.WSGIServer(('127.0.0.1', local.ipc_port.value), Router(CommandsProcessor()))
coroutine.spawn(self.server.serve_forever)
coroutine.dispatch()
client = http.Connection('http://127.0.0.1:%s' % local.ipc_port.value)
events = []
CommandsProcessor.events = ['', 'fake', 'data: fail', 'data: null', 'data: -1', 'data: {"foo": "bar"}']
try:
for i in client.subscribe():
events.append(i)
except Exception:
pass
self.assertEqual(
[-1, {'foo': 'bar'}],
events)
events = []
CommandsProcessor.events = ['', 'fake', 'data: fail', 'data: null', 'data: -1', 'data: {"foo": "bar"}']
subscription = client.subscribe()
try:
while select.select([subscription.fileno()], [], []):
events.append(subscription.pull())
assert False
except Exception:
pass
self.assertEqual(
[None, None, None, None, -1, {'foo': 'bar'}],
events)
def test_call(self):
class Commands(object):
@route('FOO', [None, None], cmd='f1', mime_type='application/json')
def f1(self, request):
return request.path
self.server = coroutine.WSGIServer(('127.0.0.1', local.ipc_port.value), Router(Commands()))
coroutine.spawn(self.server.serve_forever)
coroutine.dispatch()
conn = http.Connection('http://127.0.0.1:%s' % local.ipc_port.value)
request = Request({
'REQUEST_METHOD': 'FOO',
'PATH_INFO': '/foo/bar',
'QUERY_STRING': 'cmd=f1',
})
self.assertEqual(['foo', 'bar'], conn.call(request))
self.assertEqual(['foo', 'bar'], conn.call(Request(method='FOO', path=['foo', 'bar'], cmd='f1')))
def test_call_ReturnStream(self):
class Commands(object):
@route('GET', cmd='f1', mime_type='application/json')
def f1(self):
yield json.dumps('result')
@route('GET', cmd='f2', mime_type='foo/bar')
def f2(self):
yield json.dumps('result')
self.server = coroutine.WSGIServer(('127.0.0.1', local.ipc_port.value), Router(Commands()))
coroutine.spawn(self.server.serve_forever)
coroutine.dispatch()
client = http.Connection('http://127.0.0.1:%s' % local.ipc_port.value)
request = Request({
'REQUEST_METHOD': 'GET',
'PATH_INFO': '/',
'QUERY_STRING': 'cmd=f1',
})
self.assertEqual('result', client.call(request))
request = Request({
'REQUEST_METHOD': 'GET',
'PATH_INFO': '/',
'QUERY_STRING': 'cmd=f2',
})
self.assertEqual('result', json.load(client.call(request)))
if __name__ == '__main__':
tests.main()
|
