1import socket 2 3import pytest 4from unit.control import TestControl 5 6 7class TestConfiguration(TestControl): 8 prerequisites = {'modules': {'python': 'any'}} 9 10 def try_addr(self, addr): 11 return self.conf( 12 { 13 "listeners": {addr: {"pass": "routes"}}, 14 "routes": [{"action": {"return": 200}}], 15 "applications": {}, 16 } 17 ) 18 19 def test_json_empty(self): 20 assert 'error' in self.conf(''), 'empty' 21 22 def test_json_leading_zero(self): 23 assert 'error' in self.conf('00'), 'leading zero' 24 25 def test_json_unicode(self): 26 assert 'success' in self.conf( 27 u""" 28 { 29 "ap\u0070": { 30 "type": "\u0070ython", 31 "processes": { "spare": 0 }, 32 "path": "\u002Fapp", 33 "module": "wsgi" 34 } 35 } 36 """, 37 'applications', 38 ), 'unicode' 39 40 assert self.conf_get('applications') == { 41 "app": { 42 "type": "python", 43 "processes": {"spare": 0}, 44 "path": "/app", 45 "module": "wsgi", 46 } 47 }, 'unicode get' 48 49 def test_json_unicode_2(self): 50 assert 'success' in self.conf( 51 { 52 "приложение": { 53 "type": "python", 54 "processes": {"spare": 0}, 55 "path": "/app", 56 "module": "wsgi", 57 } 58 }, 59 'applications', 60 ), 'unicode 2' 61 62 assert 'приложение' in self.conf_get('applications'), 'unicode 2 get' 63 64 def test_json_unicode_number(self): 65 assert 'success' in self.conf( 66 u""" 67 { 68 "app": { 69 "type": "python", 70 "processes": { "spare": \u0030 }, 71 "path": "/app", 72 "module": "wsgi" 73 } 74 } 75 """, 76 'applications', 77 ), 'unicode number' 78 79 def test_json_utf8_bom(self): 80 assert 'success' in self.conf( 81 b"""\xEF\xBB\xBF 82 { 83 "app": { 84 "type": "python", 85 "processes": {"spare": 0}, 86 "path": "/app", 87 "module": "wsgi" 88 } 89 } 90 """, 91 'applications', 92 ), 'UTF-8 BOM' 93 94 def test_json_comment_single_line(self): 95 assert 'success' in self.conf( 96 b""" 97 // this is bridge 98 { 99 "//app": { 100 "type": "python", // end line 101 "processes": {"spare": 0}, 102 // inside of block 103 "path": "/app", 104 "module": "wsgi" 105 } 106 // double // 107 } 108 // end of json \xEF\t 109 """, 110 'applications', 111 ), 'single line comments' 112 113 def test_json_comment_multi_line(self): 114 assert 'success' in self.conf( 115 b""" 116 /* this is bridge */ 117 { 118 "/*app": { 119 /** 120 * multiple lines 121 **/ 122 "type": "python", 123 "processes": /* inline */ {"spare": 0}, 124 "path": "/app", 125 "module": "wsgi" 126 /* 127 // end of block */ 128 } 129 /* blah * / blah /* blah */ 130 } 131 /* end of json \xEF\t\b */ 132 """, 133 'applications', 134 ), 'multi line comments' 135 136 def test_json_comment_invalid(self): 137 assert 'error' in self.conf(b'/{}', 'applications'), 'slash' 138 assert 'error' in self.conf(b'//{}', 'applications'), 'comment' 139 assert 'error' in self.conf(b'{} /', 'applications'), 'slash end' 140 assert 'error' in self.conf(b'/*{}', 'applications'), 'slash star' 141 assert 'error' in self.conf(b'{} /*', 'applications'), 'slash star end' 142 143 def test_applications_open_brace(self): 144 assert 'error' in self.conf('{', 'applications'), 'open brace' 145 146 def test_applications_string(self): 147 assert 'error' in self.conf('"{}"', 'applications'), 'string' 148 149 @pytest.mark.skip('not yet, unsafe') 150 def test_applications_type_only(self): 151 assert 'error' in self.conf( 152 {"app": {"type": "python"}}, 'applications' 153 ), 'type only' 154 155 def test_applications_miss_quote(self): 156 assert 'error' in self.conf( 157 """ 158 { 159 app": { 160 "type": "python", 161 "processes": { "spare": 0 }, 162 "path": "/app", 163 "module": "wsgi" 164 } 165 } 166 """, 167 'applications', 168 ), 'miss quote' 169 170 def test_applications_miss_colon(self): 171 assert 'error' in self.conf( 172 """ 173 { 174 "app" { 175 "type": "python", 176 "processes": { "spare": 0 }, 177 "path": "/app", 178 "module": "wsgi" 179 } 180 } 181 """, 182 'applications', 183 ), 'miss colon' 184 185 def test_applications_miss_comma(self): 186 assert 'error' in self.conf( 187 """ 188 { 189 "app": { 190 "type": "python" 191 "processes": { "spare": 0 }, 192 "path": "/app", 193 "module": "wsgi" 194 } 195 } 196 """, 197 'applications', 198 ), 'miss comma' 199 200 def test_applications_skip_spaces(self): 201 assert 'success' in self.conf( 202 b'{ \n\r\t}', 'applications' 203 ), 'skip spaces' 204 205 def test_applications_relative_path(self): 206 assert 'success' in self.conf( 207 { 208 "app": { 209 "type": "python", 210 "processes": {"spare": 0}, 211 "path": "../app", 212 "module": "wsgi", 213 } 214 }, 215 'applications', 216 ), 'relative path' 217 218 @pytest.mark.skip('not yet, unsafe') 219 def test_listeners_empty(self): 220 assert 'error' in self.conf( 221 {"*:7080": {}}, 'listeners' 222 ), 'listener empty' 223 224 def test_listeners_no_app(self): 225 assert 'error' in self.conf( 226 {"*:7080": {"pass": "applications/app"}}, 'listeners' 227 ), 'listeners no app' 228 229 def test_listeners_addr(self): 230 assert 'success' in self.try_addr("*:7080"), 'wildcard' 231 assert 'success' in self.try_addr("127.0.0.1:7081"), 'explicit' 232 assert 'success' in self.try_addr("[::1]:7082"), 'explicit ipv6' 233 234 def test_listeners_addr_error(self): 235 assert 'error' in self.try_addr("127.0.0.1"), 'no port' 236 237 def test_listeners_addr_error_2(self, skip_alert): 238 skip_alert(r'bind.*failed', r'failed to apply new conf') 239 240 assert 'error' in self.try_addr( 241 "[f607:7403:1e4b:6c66:33b2:843f:2517:da27]:7080" 242 ) 243 244 def test_listeners_port_release(self): 245 for i in range(10): 246 fail = False 247 with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: 248 s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 249 250 self.conf( 251 { 252 "listeners": {"127.0.0.1:7080": {"pass": "routes"}}, 253 "routes": [], 254 } 255 ) 256 257 resp = self.conf({"listeners": {}, "applications": {}}) 258 259 try: 260 s.bind(('127.0.0.1', 7080)) 261 s.listen() 262 263 except OSError: 264 fail = True 265 266 if fail: 267 pytest.fail('cannot bind or listen to the address') 268 269 assert 'success' in resp, 'port release' 270 271 def test_json_application_name_large(self): 272 name = "X" * 1024 * 1024 273 274 assert 'success' in self.conf( 275 { 276 "listeners": {"*:7080": {"pass": "applications/" + name}}, 277 "applications": { 278 name: { 279 "type": "python", 280 "processes": {"spare": 0}, 281 "path": "/app", 282 "module": "wsgi", 283 } 284 }, 285 } 286 ) 287 288 @pytest.mark.skip('not yet') 289 def test_json_application_many(self): 290 apps = 999 291 292 conf = { 293 "applications": { 294 "app-" 295 + str(a): { 296 "type": "python", 297 "processes": {"spare": 0}, 298 "path": "/app", 299 "module": "wsgi", 300 } 301 for a in range(apps) 302 }, 303 "listeners": { 304 "*:" + str(7000 + a): {"pass": "applications/app-" + str(a)} 305 for a in range(apps) 306 }, 307 } 308 309 assert 'success' in self.conf(conf) 310 311 def test_json_application_many2(self): 312 conf = { 313 "applications": { 314 "app-" 315 + str(a): { 316 "type": "python", 317 "processes": {"spare": 0}, 318 "path": "/app", 319 "module": "wsgi", 320 } 321 # Larger number of applications can cause test fail with default 322 # open files limit due to the lack of file descriptors. 323 for a in range(100) 324 }, 325 "listeners": {"*:7080": {"pass": "applications/app-1"}}, 326 } 327 328 assert 'success' in self.conf(conf) 329 330 def test_unprivileged_user_error(self, is_su, skip_alert): 331 skip_alert(r'cannot set user "root"', r'failed to apply new conf') 332 if is_su: 333 pytest.skip('unprivileged tests') 334 335 assert 'error' in self.conf( 336 { 337 "app": { 338 "type": "external", 339 "processes": 1, 340 "executable": "/app", 341 "user": "root", 342 } 343 }, 344 'applications', 345 ), 'setting user' 346