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
|
#!/usr/bin/env python
import pytest
from circuits import Component
from circuits.net.events import connect, write
from circuits.net.sockets import TCPClient
from circuits.web import Controller
from circuits.web.client import parse_url
class Client(Component):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._buffer = []
self.done = False
def read(self, data):
self._buffer.append(data)
if data.find(b'\r\n') != -1:
self.done = True
def buffer(self):
return b''.join(self._buffer)
class Root(Controller):
def index(self):
return 'Hello World!'
def test(webapp):
transport = TCPClient()
client = Client()
client += transport
client.start()
host, port, _resource, _secure = parse_url(webapp.server.http.base)
client.fire(connect(host, port))
assert pytest.wait_for(transport, 'connected')
client.fire(write(b'GET / HTTP/1.1\r\n'))
client.fire(write(b'Host: localhost\r\n'))
client.fire(write(b'Content-Type: text/plain\r\n\r\n'))
assert pytest.wait_for(client, 'done')
client.stop()
ss = client.buffer().decode('utf-8')
s = ss.split('\r\n')[0]
assert s == 'HTTP/1.1 200 OK', ss
def test_http_1_0(webapp):
transport = TCPClient()
client = Client()
client += transport
client.start()
host, port, _resource, _secure = parse_url(webapp.server.http.base)
client.fire(connect(host, port))
assert pytest.wait_for(transport, 'connected')
client.fire(write(b'GET / HTTP/1.0\r\n\r\n'))
assert pytest.wait_for(client, 'done')
client.stop()
s = client.buffer().decode('utf-8').split('\r\n')[0]
assert s == 'HTTP/1.0 200 OK'
def test_http_1_1_no_host_headers(webapp):
transport = TCPClient()
client = Client()
client += transport
client.start()
host, port, _resource, _secure = parse_url(webapp.server.http.base)
client.fire(connect(host, port))
assert pytest.wait_for(transport, 'connected')
client.fire(write(b'GET / HTTP/1.1\r\n\r\n'))
assert pytest.wait_for(client, 'done')
client.stop()
s = client.buffer().decode('utf-8').split('\r\n')[0]
assert s == 'HTTP/1.1 400 Bad Request'
|
