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