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