1import unittest 2import unit 3 4class TestUnitConfiguration(unit.TestUnitControl): 5 6 def setUpClass(): 7 unit.TestUnit().check_modules('python') 8 9 def test_json_empty(self): 10 self.assertIn('error', self.conf(''), 'empty') 11 12 def test_json_leading_zero(self): 13 self.assertIn('error', self.conf('00'), 'leading zero') 14 15 def test_json_unicode(self): 16 self.assertIn('success', self.conf(b""" 17 { 18 "ap\u0070": { 19 "type": "\u0070ython", 20 "processes": { "spare": 0 }, 21 "path": "\u002Fapp", 22 "module": "wsgi" 23 } 24 } 25 """, 'applications'), 'unicode') 26 27 self.assertDictEqual(self.conf_get('applications'), { 28 "app": { 29 "type": "python", 30 "processes": { "spare": 0 }, 31 "path": "/app", 32 "module": "wsgi" 33 } 34 }, 'unicode get') 35 36 def test_json_unicode_2(self): 37 self.assertIn('success', self.conf({ 38 "приложение": { 39 "type": "python", 40 "processes": { "spare": 0 }, 41 "path": "/app", 42 "module": "wsgi" 43 } 44 }, 'applications'), 'unicode 2') 45 46 self.assertIn('приложение', self.conf_get('applications'), 47 'unicode 2 get') 48 49 def test_json_unicode_number(self): 50 self.assertIn('error', self.conf(b""" 51 { 52 "app": { 53 "type": "python", 54 "processes": { "spare": \u0030 }, 55 "path": "/app", 56 "module": "wsgi" 57 } 58 } 59 """, 'applications'), 'unicode number') 60 61 def test_applications_open_brace(self): 62 self.assertIn('error', self.conf('{', 'applications'), 'open brace') 63 64 def test_applications_string(self): 65 self.assertIn('error', self.conf('"{}"', 'applications'), 'string') 66 67 def test_applications_type_only(self): 68 self.skip_alerts.extend([ 69 r'python module is empty', 70 r'failed to apply new conf', 71 r'process \d+ exited on signal' 72 ]) 73 74 self.assertIn('error', self.conf({ 75 "app": { 76 "type": "python" 77 } 78 }, 'applications'), 'type only') 79 80 def test_applications_miss_quote(self): 81 self.assertIn('error', self.conf(""" 82 { 83 app": { 84 "type": "python", 85 "processes": { "spare": 0 }, 86 "path": "/app", 87 "module": "wsgi" 88 } 89 } 90 """, 'applications'), 'miss quote') 91 92 def test_applications_miss_colon(self): 93 self.assertIn('error', self.conf(""" 94 { 95 "app" { 96 "type": "python", 97 "processes": { "spare": 0 }, 98 "path": "/app", 99 "module": "wsgi" 100 } 101 } 102 """, 'applications'), 'miss colon') 103 104 def test_applications_miss_comma(self): 105 self.assertIn('error', self.conf(""" 106 { 107 "app": { 108 "type": "python" 109 "processes": { "spare": 0 }, 110 "path": "/app", 111 "module": "wsgi" 112 } 113 } 114 """, 'applications'), 'miss comma') 115 116 def test_applications_skip_spaces(self): 117 self.assertIn('success', self.conf(b'{ \n\r\t}', 'applications'), 118 'skip spaces') 119 120 def test_applications_relative_path(self): 121 self.assertIn('success', self.conf({ 122 "app": { 123 "type": "python", 124 "processes": { "spare": 0 }, 125 "path": "../app", 126 "module": "wsgi" 127 } 128 }, 'applications'), 'relative path') 129 130 @unittest.expectedFailure 131 def test_listeners_empty(self): 132 self.skip_sanitizer = True 133 self.skip_alerts.extend([ 134 r'failed to apply previous configuration', 135 r'process \d+ exited on signal' 136 ]) 137 138 self.assertIn('error', self.conf({"*:7080":{}}, 'listeners'), 139 'listener empty') 140 141 def test_listeners_no_app(self): 142 self.assertIn('error', self.conf({"*:7080":{"application":"app"}}, 143 'listeners'), 'listeners no app') 144 145 def test_listeners_wildcard(self): 146 self.assertIn('success', self.conf({ 147 "listeners": { 148 "*:7080": { 149 "application":"app" 150 } 151 }, 152 "applications": { 153 "app": { 154 "type": "python", 155 "processes": { "spare": 0 }, 156 "path": "/app", 157 "module": "wsgi" 158 } 159 } 160 }), 'listeners wildcard') 161 162 def test_listeners_explicit(self): 163 self.assertIn('success', self.conf({ 164 "listeners": { 165 "127.0.0.1:7080": { 166 "application":"app" 167 } 168 }, 169 "applications": { 170 "app": { 171 "type": "python", 172 "processes": { "spare": 0 }, 173 "path": "/app", 174 "module": "wsgi" 175 } 176 } 177 }), 'explicit') 178 179 def test_listeners_explicit_ipv6(self): 180 self.assertIn('success', self.conf({ 181 "listeners": { 182 "[::1]:7080": { 183 "application":"app" 184 } 185 }, 186 "applications": { 187 "app": { 188 "type": "python", 189 "processes": { "spare": 0 }, 190 "path": "/app", 191 "module": "wsgi" 192 } 193 } 194 }), 'explicit ipv6') 195 196 def test_listeners_no_port(self): 197 self.skip_alerts.extend([ 198 r'invalid listener "127\.0\.0\.1"', 199 r'failed to apply new conf', 200 r'process \d+ exited on signal' 201 ]) 202 203 self.assertIn('error', self.conf({ 204 "listeners": { 205 "127.0.0.1": { 206 "application":"app" 207 } 208 }, 209 "applications": { 210 "app": { 211 "type": "python", 212 "processes": { "spare": 0 }, 213 "path": "/app", 214 "module": "wsgi" 215 } 216 } 217 }), 'no port') 218 219 @unittest.expectedFailure 220 def test_json_application_name_large(self): 221 self.skip_alerts.append(r'epoll_ctl.+failed') 222 name = "X" * 1024 * 1024 223 224 self.assertIn('success', self.conf({ 225 "listeners": { 226 "*:7080": { 227 "application": name 228 } 229 }, 230 "applications": { 231 name: { 232 "type": "python", 233 "processes": { "spare": 0 }, 234 "path": "/app", 235 "module": "wsgi" 236 } 237 } 238 })) 239 240 @unittest.expectedFailure 241 def test_json_application_many(self): 242 self.skip_alerts.extend([ 243 r'eventfd.+failed', 244 r'epoll.+failed', 245 r'failed to apply' 246 ]) 247 apps = 999 248 249 conf = { 250 "applications": 251 {"app-" + str(a): { 252 "type": "python", 253 "processes": { "spare": 0 }, 254 "path": "/app", 255 "module": "wsgi" 256 } for a in range(apps) 257 }, 258 "listeners": { 259 "*:" + str(7000 + a): { 260 "application": "app-" + str(a) 261 } for a in range(apps) 262 } 263 } 264 265 self.assertIn('success', self.conf(conf)) 266
| 1import unittest 2import unit 3 4class TestUnitConfiguration(unit.TestUnitControl): 5 6 def setUpClass(): 7 unit.TestUnit().check_modules('python') 8 9 def test_json_empty(self): 10 self.assertIn('error', self.conf(''), 'empty') 11 12 def test_json_leading_zero(self): 13 self.assertIn('error', self.conf('00'), 'leading zero') 14 15 def test_json_unicode(self): 16 self.assertIn('success', self.conf(b""" 17 { 18 "ap\u0070": { 19 "type": "\u0070ython", 20 "processes": { "spare": 0 }, 21 "path": "\u002Fapp", 22 "module": "wsgi" 23 } 24 } 25 """, 'applications'), 'unicode') 26 27 self.assertDictEqual(self.conf_get('applications'), { 28 "app": { 29 "type": "python", 30 "processes": { "spare": 0 }, 31 "path": "/app", 32 "module": "wsgi" 33 } 34 }, 'unicode get') 35 36 def test_json_unicode_2(self): 37 self.assertIn('success', self.conf({ 38 "приложение": { 39 "type": "python", 40 "processes": { "spare": 0 }, 41 "path": "/app", 42 "module": "wsgi" 43 } 44 }, 'applications'), 'unicode 2') 45 46 self.assertIn('приложение', self.conf_get('applications'), 47 'unicode 2 get') 48 49 def test_json_unicode_number(self): 50 self.assertIn('error', self.conf(b""" 51 { 52 "app": { 53 "type": "python", 54 "processes": { "spare": \u0030 }, 55 "path": "/app", 56 "module": "wsgi" 57 } 58 } 59 """, 'applications'), 'unicode number') 60 61 def test_applications_open_brace(self): 62 self.assertIn('error', self.conf('{', 'applications'), 'open brace') 63 64 def test_applications_string(self): 65 self.assertIn('error', self.conf('"{}"', 'applications'), 'string') 66 67 def test_applications_type_only(self): 68 self.skip_alerts.extend([ 69 r'python module is empty', 70 r'failed to apply new conf', 71 r'process \d+ exited on signal' 72 ]) 73 74 self.assertIn('error', self.conf({ 75 "app": { 76 "type": "python" 77 } 78 }, 'applications'), 'type only') 79 80 def test_applications_miss_quote(self): 81 self.assertIn('error', self.conf(""" 82 { 83 app": { 84 "type": "python", 85 "processes": { "spare": 0 }, 86 "path": "/app", 87 "module": "wsgi" 88 } 89 } 90 """, 'applications'), 'miss quote') 91 92 def test_applications_miss_colon(self): 93 self.assertIn('error', self.conf(""" 94 { 95 "app" { 96 "type": "python", 97 "processes": { "spare": 0 }, 98 "path": "/app", 99 "module": "wsgi" 100 } 101 } 102 """, 'applications'), 'miss colon') 103 104 def test_applications_miss_comma(self): 105 self.assertIn('error', self.conf(""" 106 { 107 "app": { 108 "type": "python" 109 "processes": { "spare": 0 }, 110 "path": "/app", 111 "module": "wsgi" 112 } 113 } 114 """, 'applications'), 'miss comma') 115 116 def test_applications_skip_spaces(self): 117 self.assertIn('success', self.conf(b'{ \n\r\t}', 'applications'), 118 'skip spaces') 119 120 def test_applications_relative_path(self): 121 self.assertIn('success', self.conf({ 122 "app": { 123 "type": "python", 124 "processes": { "spare": 0 }, 125 "path": "../app", 126 "module": "wsgi" 127 } 128 }, 'applications'), 'relative path') 129 130 @unittest.expectedFailure 131 def test_listeners_empty(self): 132 self.skip_sanitizer = True 133 self.skip_alerts.extend([ 134 r'failed to apply previous configuration', 135 r'process \d+ exited on signal' 136 ]) 137 138 self.assertIn('error', self.conf({"*:7080":{}}, 'listeners'), 139 'listener empty') 140 141 def test_listeners_no_app(self): 142 self.assertIn('error', self.conf({"*:7080":{"application":"app"}}, 143 'listeners'), 'listeners no app') 144 145 def test_listeners_wildcard(self): 146 self.assertIn('success', self.conf({ 147 "listeners": { 148 "*:7080": { 149 "application":"app" 150 } 151 }, 152 "applications": { 153 "app": { 154 "type": "python", 155 "processes": { "spare": 0 }, 156 "path": "/app", 157 "module": "wsgi" 158 } 159 } 160 }), 'listeners wildcard') 161 162 def test_listeners_explicit(self): 163 self.assertIn('success', self.conf({ 164 "listeners": { 165 "127.0.0.1:7080": { 166 "application":"app" 167 } 168 }, 169 "applications": { 170 "app": { 171 "type": "python", 172 "processes": { "spare": 0 }, 173 "path": "/app", 174 "module": "wsgi" 175 } 176 } 177 }), 'explicit') 178 179 def test_listeners_explicit_ipv6(self): 180 self.assertIn('success', self.conf({ 181 "listeners": { 182 "[::1]:7080": { 183 "application":"app" 184 } 185 }, 186 "applications": { 187 "app": { 188 "type": "python", 189 "processes": { "spare": 0 }, 190 "path": "/app", 191 "module": "wsgi" 192 } 193 } 194 }), 'explicit ipv6') 195 196 def test_listeners_no_port(self): 197 self.skip_alerts.extend([ 198 r'invalid listener "127\.0\.0\.1"', 199 r'failed to apply new conf', 200 r'process \d+ exited on signal' 201 ]) 202 203 self.assertIn('error', self.conf({ 204 "listeners": { 205 "127.0.0.1": { 206 "application":"app" 207 } 208 }, 209 "applications": { 210 "app": { 211 "type": "python", 212 "processes": { "spare": 0 }, 213 "path": "/app", 214 "module": "wsgi" 215 } 216 } 217 }), 'no port') 218 219 @unittest.expectedFailure 220 def test_json_application_name_large(self): 221 self.skip_alerts.append(r'epoll_ctl.+failed') 222 name = "X" * 1024 * 1024 223 224 self.assertIn('success', self.conf({ 225 "listeners": { 226 "*:7080": { 227 "application": name 228 } 229 }, 230 "applications": { 231 name: { 232 "type": "python", 233 "processes": { "spare": 0 }, 234 "path": "/app", 235 "module": "wsgi" 236 } 237 } 238 })) 239 240 @unittest.expectedFailure 241 def test_json_application_many(self): 242 self.skip_alerts.extend([ 243 r'eventfd.+failed', 244 r'epoll.+failed', 245 r'failed to apply' 246 ]) 247 apps = 999 248 249 conf = { 250 "applications": 251 {"app-" + str(a): { 252 "type": "python", 253 "processes": { "spare": 0 }, 254 "path": "/app", 255 "module": "wsgi" 256 } for a in range(apps) 257 }, 258 "listeners": { 259 "*:" + str(7000 + a): { 260 "application": "app-" + str(a) 261 } for a in range(apps) 262 } 263 } 264 265 self.assertIn('success', self.conf(conf)) 266
|