xref: /unit/test/test_return.py (revision 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})
186        check_error({"return": -1})
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()
199