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
117
118
119
120
121
122
123
124
125
126
|
import email
import io
import socket
from paste.httpserver import LimitedLengthFile, WSGIHandler, serve
class MockServer:
server_address = ('127.0.0.1', 80)
class MockSocket:
def makefile(self, mode, bufsize):
return io.StringIO()
def test_environ():
mock_socket = MockSocket()
mock_client_address = '1.2.3.4'
mock_server = MockServer()
wsgi_handler = WSGIHandler(mock_socket, mock_client_address, mock_server)
wsgi_handler.command = 'GET'
wsgi_handler.path = '/path'
wsgi_handler.request_version = 'HTTP/1.0'
wsgi_handler.headers = email.message_from_string('Host: mywebsite')
wsgi_handler.wsgi_setup()
assert wsgi_handler.wsgi_environ['HTTP_HOST'] == 'mywebsite'
def test_environ_with_multiple_values():
mock_socket = MockSocket()
mock_client_address = '1.2.3.4'
mock_server = MockServer()
wsgi_handler = WSGIHandler(mock_socket, mock_client_address, mock_server)
wsgi_handler.command = 'GET'
wsgi_handler.path = '/path'
wsgi_handler.request_version = 'HTTP/1.0'
wsgi_handler.headers = email.message_from_string('Host: host1\nHost: host2')
wsgi_handler.wsgi_setup()
assert wsgi_handler.wsgi_environ['HTTP_HOST'] == 'host1,host2'
def test_limited_length_file():
backing = io.BytesIO(b'0123456789')
f = LimitedLengthFile(backing, 9)
assert f.tell() == 0
assert f.read() == b'012345678'
assert f.tell() == 9
assert f.read() == b''
def test_limited_length_file_tell_on_socket():
backing_read, backing_write = socket.socketpair()
f = LimitedLengthFile(backing_read.makefile('rb'), 10)
backing_write.send(b'0123456789')
backing_write.close()
assert f.tell() == 0
assert f.read(1) == b'0'
assert f.tell() == 1
assert f.read() == b'123456789'
assert f.tell() == 10
backing_read.close()
def test_address_family_v4():
#ipv4
app = None
host = '127.0.0.1'
port = '9090'
svr = serve(app, host=host, port=port, start_loop=False, use_threadpool=False)
af = svr.address_family
addr = svr.server_address
p = svr.server_port
svr.server_close()
assert (af == socket.AF_INET)
assert (addr[0] == '127.0.0.1')
assert (str(p) == port)
def test_address_family_v4_host_and_port():
#ipv4
app = None
host = '127.0.0.1:9091'
svr = serve(app, host=host, start_loop=False, use_threadpool=False)
af = svr.address_family
addr = svr.server_address
p = svr.server_port
svr.server_close()
assert (af == socket.AF_INET)
assert (addr[0] == '127.0.0.1')
assert (str(p) == '9091')
def test_address_family_v6():
#ipv6
app = None
host = '[::1]'
port = '9090'
try:
svr = serve(app, host=host, port=port, start_loop=False, use_threadpool=False)
af = svr.address_family
addr = svr.server_address
p = svr.server_port
svr.server_close()
assert (af == socket.AF_INET6)
assert (addr[0] == '::1')
assert (str(p) == port)
except (socket.error, OSError) as err:
# v6 support not available in this OS, pass the test
assert True
|
