test_return.py (1441:b2caecb224f7) test_return.py (1443:433622b9cc55)
1import re
2import unittest
3from unit.applications.proto import TestApplicationProto
4
5
6class TestReturn(TestApplicationProto):
7 prerequisites = {}
8
9 def setUp(self):
10 super().setUp()
11
12 self._load_conf(
13 {
14 "listeners": {"*:7080": {"pass": "routes"}},
15 "routes": [{"action": {"return": 200}}],
16 "applications": {},
17 }
18 )
19
20 def get_resps_sc(self, req=10):
21 to_send = b"""GET / HTTP/1.1
22Host: localhost
23
24""" * (
25 req - 1
26 )
27
28 to_send += b"""GET / HTTP/1.1
29Host: localhost
30Connection: close
31
32"""
33
34 return self.http(to_send, raw_resp=True, raw=True)
35
36 def test_return(self):
37 resp = self.get()
38 self.assertEqual(resp['status'], 200)
39 self.assertIn('Server', resp['headers'])
40 self.assertIn('Date', resp['headers'])
41 self.assertEqual(resp['headers']['Content-Length'], '0')
42 self.assertEqual(resp['headers']['Connection'], 'close')
43 self.assertEqual(resp['body'], '', 'body')
44
45 resp = self.post(body='blah')
46 self.assertEqual(resp['status'], 200)
47 self.assertEqual(resp['body'], '', 'body')
48
49 resp = self.get_resps_sc()
50 self.assertEqual(len(re.findall('200 OK', resp)), 10)
51 self.assertEqual(len(re.findall('Connection:', resp)), 1)
52 self.assertEqual(len(re.findall('Connection: close', resp)), 1)
53
54 resp = self.get(http_10=True)
55 self.assertEqual(resp['status'], 200)
56 self.assertIn('Server', resp['headers'])
57 self.assertIn('Date', resp['headers'])
58 self.assertEqual(resp['headers']['Content-Length'], '0')
59 self.assertNotIn('Connection', resp['headers'])
60 self.assertEqual(resp['body'], '', 'body')
61
62 def test_return_update(self):
63 self.assertIn('success', self.conf('0', 'routes/0/action/return'))
64
65 resp = self.get()
66 self.assertEqual(resp['status'], 0)
67 self.assertEqual(resp['body'], '')
68
69 self.assertIn('success', self.conf('404', 'routes/0/action/return'))
70
71 resp = self.get()
72 self.assertEqual(resp['status'], 404)
73 self.assertNotEqual(resp['body'], '')
74
75 self.assertIn('success', self.conf('598', 'routes/0/action/return'))
76
77 resp = self.get()
78 self.assertEqual(resp['status'], 598)
79 self.assertNotEqual(resp['body'], '')
80
81 self.assertIn('success', self.conf('999', 'routes/0/action/return'))
82
83 resp = self.get()
84 self.assertEqual(resp['status'], 999)
85 self.assertEqual(resp['body'], '')
86
87 def test_return_location(self):
88 reserved = ":/?#[]@!$&'()*+,;="
89 unreserved = ("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
90 "0123456789-._~")
91 unsafe = " \"%<>\\^`{|}"
92 unsafe_enc = "%20%22%25%3C%3E%5C%5E%60%7B%7C%7D"
93
94 def check_location(location, expect=None):
95 if expect is None:
96 expect = location
97
98 self.assertIn(
99 'success',
100 self.conf(
101 {"return": 301, "location": location}, 'routes/0/action'
102 ),
103 'configure location'
104 )
105
106 self.assertEqual(self.get()['headers']['Location'], expect)
107
108 # FAIL: can't specify empty header value.
109 # check_location("")
110
111 check_location(reserved)
112
113 # After first "?" all other "?" encoded.
114 check_location("/?" + reserved, "/?:/%3F#[]@!$&'()*+,;=")
115 check_location("???", "?%3F%3F")
116
117 # After first "#" all other "?" or "#" encoded.
118 check_location("/#" + reserved, "/#:/%3F%23[]@!$&'()*+,;=")
119 check_location("##?#?", "#%23%3F%23%3F")
120
121 # After first "?" next "#" not encoded.
122 check_location("/?#" + reserved, "/?#:/%3F%23[]@!$&'()*+,;=")
123 check_location("??##", "?%3F#%23")
124 check_location("/?##?", "/?#%23%3F")
125
126 # Unreserved never encoded.
127 check_location(unreserved)
128 check_location("/" + unreserved + "?" + unreserved + "#" + unreserved)
129
130 # Unsafe always encoded.
131 check_location(unsafe, unsafe_enc)
132 check_location("?" + unsafe, "?" + unsafe_enc)
133 check_location("#" + unsafe, "#" + unsafe_enc)
134
135 # %00-%20 and %7F-%FF always encoded.
136 check_location(u"\u0000\u0018\u001F\u0020\u0021", "%00%18%1F%20!")
137 check_location(u"\u007F\u0080н\u20BD", "%7F%C2%80%D0%BD%E2%82%BD")
138
139 # Encoded string detection. If at least one char need to be encoded
140 # then whole string will be encoded.
141 check_location("%20")
142 check_location("/%20?%20#%20")
143 check_location(" %20", "%20%2520")
144 check_location("%20 ", "%2520%20")
145 check_location("/%20?%20#%20 ", "/%2520?%2520#%2520%20")
146
147 def test_return_location_edit(self):
148 self.assertIn(
149 'success',
150 self.conf(
151 {"return": 302, "location": "blah"}, 'routes/0/action'
152 ),
153 'configure init location'
154 )
155 self.assertEqual(self.get()['headers']['Location'], 'blah')
156
157 self.assertIn(
158 'success',
159 self.conf_delete('routes/0/action/location'),
160 'location delete'
161 )
162 self.assertNotIn('Location', self.get()['headers'])
163
164 self.assertIn(
165 'success',
166 self.conf('"blah"', 'routes/0/action/location'),
167 'location restore'
168 )
169 self.assertEqual(self.get()['headers']['Location'], 'blah')
170
171 self.assertIn(
172 'error',
173 self.conf_post('"blah"', 'routes/0/action/location'),
174 'location method not allowed'
175 )
176 self.assertEqual(self.get()['headers']['Location'], 'blah')
177
178 def test_return_invalid(self):
179 def check_error(conf):
180 self.assertIn('error', self.conf(conf, 'routes/0/action'))
181
182 check_error({"return": "200"})
183 check_error({"return": []})
184 check_error({"return": 80.1})
185 check_error({"return": 1000})
1import re
2import unittest
3from unit.applications.proto import TestApplicationProto
4
5
6class TestReturn(TestApplicationProto):
7 prerequisites = {}
8
9 def setUp(self):
10 super().setUp()
11
12 self._load_conf(
13 {
14 "listeners": {"*:7080": {"pass": "routes"}},
15 "routes": [{"action": {"return": 200}}],
16 "applications": {},
17 }
18 )
19
20 def get_resps_sc(self, req=10):
21 to_send = b"""GET / HTTP/1.1
22Host: localhost
23
24""" * (
25 req - 1
26 )
27
28 to_send += b"""GET / HTTP/1.1
29Host: localhost
30Connection: close
31
32"""
33
34 return self.http(to_send, raw_resp=True, raw=True)
35
36 def test_return(self):
37 resp = self.get()
38 self.assertEqual(resp['status'], 200)
39 self.assertIn('Server', resp['headers'])
40 self.assertIn('Date', resp['headers'])
41 self.assertEqual(resp['headers']['Content-Length'], '0')
42 self.assertEqual(resp['headers']['Connection'], 'close')
43 self.assertEqual(resp['body'], '', 'body')
44
45 resp = self.post(body='blah')
46 self.assertEqual(resp['status'], 200)
47 self.assertEqual(resp['body'], '', 'body')
48
49 resp = self.get_resps_sc()
50 self.assertEqual(len(re.findall('200 OK', resp)), 10)
51 self.assertEqual(len(re.findall('Connection:', resp)), 1)
52 self.assertEqual(len(re.findall('Connection: close', resp)), 1)
53
54 resp = self.get(http_10=True)
55 self.assertEqual(resp['status'], 200)
56 self.assertIn('Server', resp['headers'])
57 self.assertIn('Date', resp['headers'])
58 self.assertEqual(resp['headers']['Content-Length'], '0')
59 self.assertNotIn('Connection', resp['headers'])
60 self.assertEqual(resp['body'], '', 'body')
61
62 def test_return_update(self):
63 self.assertIn('success', self.conf('0', 'routes/0/action/return'))
64
65 resp = self.get()
66 self.assertEqual(resp['status'], 0)
67 self.assertEqual(resp['body'], '')
68
69 self.assertIn('success', self.conf('404', 'routes/0/action/return'))
70
71 resp = self.get()
72 self.assertEqual(resp['status'], 404)
73 self.assertNotEqual(resp['body'], '')
74
75 self.assertIn('success', self.conf('598', 'routes/0/action/return'))
76
77 resp = self.get()
78 self.assertEqual(resp['status'], 598)
79 self.assertNotEqual(resp['body'], '')
80
81 self.assertIn('success', self.conf('999', 'routes/0/action/return'))
82
83 resp = self.get()
84 self.assertEqual(resp['status'], 999)
85 self.assertEqual(resp['body'], '')
86
87 def test_return_location(self):
88 reserved = ":/?#[]@!$&'()*+,;="
89 unreserved = ("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"
90 "0123456789-._~")
91 unsafe = " \"%<>\\^`{|}"
92 unsafe_enc = "%20%22%25%3C%3E%5C%5E%60%7B%7C%7D"
93
94 def check_location(location, expect=None):
95 if expect is None:
96 expect = location
97
98 self.assertIn(
99 'success',
100 self.conf(
101 {"return": 301, "location": location}, 'routes/0/action'
102 ),
103 'configure location'
104 )
105
106 self.assertEqual(self.get()['headers']['Location'], expect)
107
108 # FAIL: can't specify empty header value.
109 # check_location("")
110
111 check_location(reserved)
112
113 # After first "?" all other "?" encoded.
114 check_location("/?" + reserved, "/?:/%3F#[]@!$&'()*+,;=")
115 check_location("???", "?%3F%3F")
116
117 # After first "#" all other "?" or "#" encoded.
118 check_location("/#" + reserved, "/#:/%3F%23[]@!$&'()*+,;=")
119 check_location("##?#?", "#%23%3F%23%3F")
120
121 # After first "?" next "#" not encoded.
122 check_location("/?#" + reserved, "/?#:/%3F%23[]@!$&'()*+,;=")
123 check_location("??##", "?%3F#%23")
124 check_location("/?##?", "/?#%23%3F")
125
126 # Unreserved never encoded.
127 check_location(unreserved)
128 check_location("/" + unreserved + "?" + unreserved + "#" + unreserved)
129
130 # Unsafe always encoded.
131 check_location(unsafe, unsafe_enc)
132 check_location("?" + unsafe, "?" + unsafe_enc)
133 check_location("#" + unsafe, "#" + unsafe_enc)
134
135 # %00-%20 and %7F-%FF always encoded.
136 check_location(u"\u0000\u0018\u001F\u0020\u0021", "%00%18%1F%20!")
137 check_location(u"\u007F\u0080н\u20BD", "%7F%C2%80%D0%BD%E2%82%BD")
138
139 # Encoded string detection. If at least one char need to be encoded
140 # then whole string will be encoded.
141 check_location("%20")
142 check_location("/%20?%20#%20")
143 check_location(" %20", "%20%2520")
144 check_location("%20 ", "%2520%20")
145 check_location("/%20?%20#%20 ", "/%2520?%2520#%2520%20")
146
147 def test_return_location_edit(self):
148 self.assertIn(
149 'success',
150 self.conf(
151 {"return": 302, "location": "blah"}, 'routes/0/action'
152 ),
153 'configure init location'
154 )
155 self.assertEqual(self.get()['headers']['Location'], 'blah')
156
157 self.assertIn(
158 'success',
159 self.conf_delete('routes/0/action/location'),
160 'location delete'
161 )
162 self.assertNotIn('Location', self.get()['headers'])
163
164 self.assertIn(
165 'success',
166 self.conf('"blah"', 'routes/0/action/location'),
167 'location restore'
168 )
169 self.assertEqual(self.get()['headers']['Location'], 'blah')
170
171 self.assertIn(
172 'error',
173 self.conf_post('"blah"', 'routes/0/action/location'),
174 'location method not allowed'
175 )
176 self.assertEqual(self.get()['headers']['Location'], 'blah')
177
178 def test_return_invalid(self):
179 def check_error(conf):
180 self.assertIn('error', self.conf(conf, 'routes/0/action'))
181
182 check_error({"return": "200"})
183 check_error({"return": []})
184 check_error({"return": 80.1})
185 check_error({"return": 1000})
186 check_error({"return": -1})
186 check_error({"return": 200, "share": "/blah"})
187
188 self.assertIn(
189 'error', self.conf('001', 'routes/0/action/return'), 'leading zero'
190 )
191
192 check_error({"return": 301, "location": 0})
193 check_error({"return": 301, "location": []})
194
195
196if __name__ == '__main__':
197 TestReturn.main()
187 check_error({"return": 200, "share": "/blah"})
188
189 self.assertIn(
190 'error', self.conf('001', 'routes/0/action/return'), 'leading zero'
191 )
192
193 check_error({"return": 301, "location": 0})
194 check_error({"return": 301, "location": []})
195
196
197if __name__ == '__main__':
198 TestReturn.main()