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