diff options
author | Andrey Zelenkov <zelenkov@nginx.com> | 2019-03-26 23:38:30 +0300 |
---|---|---|
committer | Andrey Zelenkov <zelenkov@nginx.com> | 2019-03-26 23:38:30 +0300 |
commit | 281899fcef10eaf815d90958d49243c5060ffac0 (patch) | |
tree | 258d01fb832a4253815db41673c9fcc73f128e6b | |
parent | 3d7a47c9acb1120f90225c833fd56cffeb99c2cd (diff) | |
download | unit-281899fcef10eaf815d90958d49243c5060ffac0.tar.gz unit-281899fcef10eaf815d90958d49243c5060ffac0.tar.bz2 |
Tests: style.
-rw-r--r-- | test/test_access_log.py | 151 | ||||
-rw-r--r-- | test/test_configuration.py | 400 | ||||
-rw-r--r-- | test/test_go_application.py | 144 | ||||
-rw-r--r-- | test/test_http_header.py | 476 | ||||
-rw-r--r-- | test/test_java_application.py | 1178 | ||||
-rw-r--r-- | test/test_node_application.py | 340 | ||||
-rw-r--r-- | test/test_perl_application.py | 157 | ||||
-rw-r--r-- | test/test_php_application.py | 330 | ||||
-rw-r--r-- | test/test_php_basic.py | 141 | ||||
-rw-r--r-- | test/test_python_application.py | 380 | ||||
-rw-r--r-- | test/test_python_basic.py | 148 | ||||
-rw-r--r-- | test/test_python_environment.py | 215 | ||||
-rw-r--r-- | test/test_python_procman.py | 189 | ||||
-rw-r--r-- | test/test_routing.py | 990 | ||||
-rw-r--r-- | test/test_ruby_application.py | 204 | ||||
-rw-r--r-- | test/test_settings.py | 143 | ||||
-rw-r--r-- | test/test_tls.py | 469 | ||||
-rw-r--r-- | test/unit.py | 484 |
18 files changed, 4159 insertions, 2380 deletions
diff --git a/test/test_access_log.py b/test/test_access_log.py index d6741c28..e0edd9fc 100644 --- a/test/test_access_log.py +++ b/test/test_access_log.py @@ -5,8 +5,8 @@ from subprocess import call import unittest import unit -class TestUnitAccessLog(unit.TestUnitApplicationPython): +class TestUnitAccessLog(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') @@ -22,34 +22,45 @@ class TestUnitAccessLog(unit.TestUnitApplicationPython): def test_access_log_keepalive(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='01234') + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='01234', + ) time.sleep(0.2) self.assertIsNotNone( - self.search_in_log(r'"POST / HTTP/1.1" 200 5'), 'keepalive 1') - - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + self.search_in_log(r'"POST / HTTP/1.1" 200 5'), 'keepalive 1' + ) + + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) time.sleep(0.2) self.stop() self.assertIsNotNone( - self.search_in_log(r'"POST / HTTP/1.1" 200 10'), 'keepalive 2') + self.search_in_log(r'"POST / HTTP/1.1" 200 10'), 'keepalive 2' + ) def test_access_log_pipeline(self): self.load('empty') - self.http(b"""GET / HTTP/1.1 + self.http( + b"""GET / HTTP/1.1 Host: localhost Referer: Referer-1 @@ -62,7 +73,10 @@ Host: localhost Referer: Referer-3 Connection: close -""", raw_resp=True, raw=True) +""", + raw_resp=True, + raw=True, + ) time.sleep(0.2) @@ -70,22 +84,21 @@ Connection: close self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-1" "-"'), - 'pipeline 1') + 'pipeline 1', + ) self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-2" "-"'), - 'pipeline 2') + 'pipeline 2', + ) self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "Referer-3" "-"'), - 'pipeline 3') + 'pipeline 3', + ) def test_access_log_ipv6(self): self.load('empty') - self.conf({ - "[::1]:7080": { - "application": "empty" - } - }, 'listeners') + self.conf({"[::1]:7080": {"application": "empty"}}, 'listeners') self.get(sock_type='ipv6') @@ -95,18 +108,17 @@ Connection: close self.assertIsNotNone( self.search_in_log( - r'::1 - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"'), 'ipv6') + r'::1 - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"' + ), + 'ipv6', + ) def test_access_log_unix(self): self.load('empty') addr = self.testdir + '/sock' - self.conf({ - "unix:" + addr: { - "application": "empty" - } - }, 'listeners') + self.conf({"unix:" + addr: {"application": "empty"}}, 'listeners') self.get(sock_type='unix', addr=addr) @@ -114,17 +126,23 @@ Connection: close self.stop() - self.assertIsNotNone(self.search_in_log( - r'unix: - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"'), 'unix') + self.assertIsNotNone( + self.search_in_log( + r'unix: - - \[.+\] "GET / HTTP/1.1" 200 0 "-" "-"' + ), + 'unix', + ) def test_access_log_referer(self): self.load('empty') - self.get(headers={ - 'Host': 'localhost', - 'Referer': 'referer-value', - 'Connection': 'close' - }) + self.get( + headers={ + 'Host': 'localhost', + 'Referer': 'referer-value', + 'Connection': 'close', + } + ) time.sleep(0.2) @@ -132,16 +150,19 @@ Connection: close self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "referer-value" "-"'), - 'referer') + 'referer', + ) def test_access_log_user_agent(self): self.load('empty') - self.get(headers={ - 'Host': 'localhost', - 'User-Agent': 'user-agent-value', - 'Connection': 'close' - }) + self.get( + headers={ + 'Host': 'localhost', + 'User-Agent': 'user-agent-value', + 'Connection': 'close', + } + ) time.sleep(0.2) @@ -149,7 +170,10 @@ Connection: close self.assertIsNotNone( self.search_in_log( - r'"GET / HTTP/1.1" 200 0 "-" "user-agent-value"'), 'user agent') + r'"GET / HTTP/1.1" 200 0 "-" "user-agent-value"' + ), + 'user agent', + ) def test_access_log_http10(self): self.load('empty') @@ -161,8 +185,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log( - r'"GET / HTTP/1.0" 200 0 "-" "-"'), 'http 1.0') + self.search_in_log(r'"GET / HTTP/1.0" 200 0 "-" "-"'), 'http 1.0' + ) def test_access_log_partial(self): self.load('empty') @@ -174,7 +198,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log(r'"GE" 400 0 "-" "-"'), 'partial') + self.search_in_log(r'"GE" 400 0 "-" "-"'), 'partial' + ) def test_access_log_partial_2(self): self.load('empty') @@ -186,7 +211,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log(r'"GET /" 400 \d+ "-" "-"'), 'partial 2') + self.search_in_log(r'"GET /" 400 \d+ "-" "-"'), 'partial 2' + ) def test_access_log_partial_3(self): self.load('empty') @@ -198,7 +224,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log(r'"GET /" 400 0 "-" "-"'), 'partial 3') + self.search_in_log(r'"GET /" 400 0 "-" "-"'), 'partial 3' + ) def test_access_log_partial_4(self): self.load('empty') @@ -210,8 +237,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log(r'"GET / HTTP/1.1" 400 0 "-" "-"'), - 'partial 4') + self.search_in_log(r'"GET / HTTP/1.1" 400 0 "-" "-"'), 'partial 4' + ) def test_access_log_partial_5(self): self.load('empty') @@ -221,7 +248,8 @@ Connection: close self.stop() self.assertIsNotNone( - self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"'), 'partial 5') + self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"'), 'partial 5' + ) def test_access_log_get_parameters(self): self.load('empty') @@ -234,8 +262,10 @@ Connection: close self.assertIsNotNone( self.search_in_log( - r'"GET /\?blah&var=val HTTP/1.1" 200 0 "-" "-"'), - 'get parameters') + r'"GET /\?blah&var=val HTTP/1.1" 200 0 "-" "-"' + ), + 'get parameters', + ) def test_access_log_delete(self): self.load('empty') @@ -265,7 +295,8 @@ Connection: close self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"', 'new.log'), - 'change') + 'change', + ) def test_access_log_reopen(self): self.load('empty') @@ -284,7 +315,8 @@ Connection: close self.assertIsNotNone( self.search_in_log(r'"GET / HTTP/1.1" 200 0 "-" "-"', 'new.log'), - 'rename new') + 'rename new', + ) self.assertFalse(os.path.isfile(log_path), 'rename old') with open(self.testdir + '/unit.pid', 'r') as f: @@ -299,10 +331,13 @@ Connection: close time.sleep(0.2) self.assertIsNone( - self.search_in_log(r'/usr1', 'new.log'), 'rename new 2') + self.search_in_log(r'/usr1', 'new.log'), 'rename new 2' + ) self.assertIsNotNone( self.search_in_log(r'"GET /usr1 HTTP/1.1" 200 0 "-" "-"'), - 'reopen 2') + 'reopen 2', + ) + if __name__ == '__main__': TestUnitAccessLog.main() diff --git a/test/test_configuration.py b/test/test_configuration.py index 20c5fc8b..4848f741 100644 --- a/test/test_configuration.py +++ b/test/test_configuration.py @@ -1,8 +1,8 @@ import unittest import unit -class TestUnitConfiguration(unit.TestUnitControl): +class TestUnitConfiguration(unit.TestUnitControl): def setUpClass(): unit.TestUnit().check_modules('python') @@ -13,7 +13,10 @@ class TestUnitConfiguration(unit.TestUnitControl): self.assertIn('error', self.conf('00'), 'leading zero') def test_json_unicode(self): - self.assertIn('success', self.conf(b""" + self.assertIn( + 'success', + self.conf( + b""" { "ap\u0070": { "type": "\u0070ython", @@ -22,32 +25,51 @@ class TestUnitConfiguration(unit.TestUnitControl): "module": "wsgi" } } - """, 'applications'), 'unicode') - - self.assertDictEqual(self.conf_get('applications'), { - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - }, 'unicode get') + """, + 'applications', + ), + 'unicode', + ) + + self.assertDictEqual( + self.conf_get('applications'), + { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, + 'unicode get', + ) def test_json_unicode_2(self): - self.assertIn('success', self.conf({ - "приложение": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - }, 'applications'), 'unicode 2') - - self.assertIn('приложение', self.conf_get('applications'), - 'unicode 2 get') + self.assertIn( + 'success', + self.conf( + { + "приложение": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, + 'applications', + ), + 'unicode 2', + ) + + self.assertIn( + 'приложение', self.conf_get('applications'), 'unicode 2 get' + ) def test_json_unicode_number(self): - self.assertIn('error', self.conf(b""" + self.assertIn( + 'error', + self.conf( + b""" { "app": { "type": "python", @@ -56,7 +78,11 @@ class TestUnitConfiguration(unit.TestUnitControl): "module": "wsgi" } } - """, 'applications'), 'unicode number') + """, + 'applications', + ), + 'unicode number', + ) def test_applications_open_brace(self): self.assertIn('error', self.conf('{', 'applications'), 'open brace') @@ -65,20 +91,25 @@ class TestUnitConfiguration(unit.TestUnitControl): self.assertIn('error', self.conf('"{}"', 'applications'), 'string') def test_applications_type_only(self): - self.skip_alerts.extend([ - r'python module is empty', - r'failed to apply new conf', - r'process \d+ exited on signal' - ]) - - self.assertIn('error', self.conf({ - "app": { - "type": "python" - } - }, 'applications'), 'type only') + self.skip_alerts.extend( + [ + r'python module is empty', + r'failed to apply new conf', + r'process \d+ exited on signal', + ] + ) + + self.assertIn( + 'error', + self.conf({"app": {"type": "python"}}, 'applications'), + 'type only', + ) def test_applications_miss_quote(self): - self.assertIn('error', self.conf(""" + self.assertIn( + 'error', + self.conf( + """ { app": { "type": "python", @@ -87,10 +118,17 @@ class TestUnitConfiguration(unit.TestUnitControl): "module": "wsgi" } } - """, 'applications'), 'miss quote') + """, + 'applications', + ), + 'miss quote', + ) def test_applications_miss_colon(self): - self.assertIn('error', self.conf(""" + self.assertIn( + 'error', + self.conf( + """ { "app" { "type": "python", @@ -99,10 +137,17 @@ class TestUnitConfiguration(unit.TestUnitControl): "module": "wsgi" } } - """, 'applications'), 'miss colon') + """, + 'applications', + ), + 'miss colon', + ) def test_applications_miss_comma(self): - self.assertIn('error', self.conf(""" + self.assertIn( + 'error', + self.conf( + """ { "app": { "type": "python" @@ -111,181 +156,206 @@ class TestUnitConfiguration(unit.TestUnitControl): "module": "wsgi" } } - """, 'applications'), 'miss comma') + """, + 'applications', + ), + 'miss comma', + ) def test_applications_skip_spaces(self): - self.assertIn('success', self.conf(b'{ \n\r\t}', 'applications'), - 'skip spaces') + self.assertIn( + 'success', self.conf(b'{ \n\r\t}', 'applications'), 'skip spaces' + ) def test_applications_relative_path(self): - self.assertIn('success', self.conf({ - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "../app", - "module": "wsgi" - } - }, 'applications'), 'relative path') + self.assertIn( + 'success', + self.conf( + { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "../app", + "module": "wsgi", + } + }, + 'applications', + ), + 'relative path', + ) @unittest.expectedFailure def test_listeners_empty(self): self.skip_sanitizer = True - self.skip_alerts.extend([ - r'failed to apply previous configuration', - r'process \d+ exited on signal' - ]) + self.skip_alerts.extend( + [ + r'failed to apply previous configuration', + r'process \d+ exited on signal', + ] + ) - self.assertIn('error', self.conf({"*:7080":{}}, 'listeners'), - 'listener empty') + self.assertIn( + 'error', self.conf({"*:7080": {}}, 'listeners'), 'listener empty' + ) def test_listeners_no_app(self): - self.assertIn('error', self.conf({"*:7080":{"application":"app"}}, - 'listeners'), 'listeners no app') + self.assertIn( + 'error', + self.conf({"*:7080": {"application": "app"}}, 'listeners'), + 'listeners no app', + ) def test_listeners_wildcard(self): - self.assertIn('success', self.conf({ - "listeners": { - "*:7080": { - "application":"app" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"*:7080": {"application": "app"}}, + "applications": { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, } - }, - "applications": { - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - } - }), 'listeners wildcard') + ), + 'listeners wildcard', + ) def test_listeners_explicit(self): - self.assertIn('success', self.conf({ - "listeners": { - "127.0.0.1:7080": { - "application":"app" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"127.0.0.1:7080": {"application": "app"}}, + "applications": { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, } - }, - "applications": { - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - } - }), 'explicit') + ), + 'explicit', + ) def test_listeners_explicit_ipv6(self): - self.assertIn('success', self.conf({ - "listeners": { - "[::1]:7080": { - "application":"app" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"[::1]:7080": {"application": "app"}}, + "applications": { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, } - }, - "applications": { - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - } - }), 'explicit ipv6') + ), + 'explicit ipv6', + ) def test_listeners_no_port(self): - self.skip_alerts.extend([ - r'invalid listener "127\.0\.0\.1"', - r'failed to apply new conf', - r'process \d+ exited on signal' - ]) - - self.assertIn('error', self.conf({ - "listeners": { - "127.0.0.1": { - "application":"app" + self.skip_alerts.extend( + [ + r'invalid listener "127\.0\.0\.1"', + r'failed to apply new conf', + r'process \d+ exited on signal', + ] + ) + + self.assertIn( + 'error', + self.conf( + { + "listeners": {"127.0.0.1": {"application": "app"}}, + "applications": { + "app": { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, } - }, - "applications": { - "app": { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" - } - } - }), 'no port') + ), + 'no port', + ) def test_json_application_name_large(self): name = "X" * 1024 * 1024 - self.assertIn('success', self.conf({ - "listeners": { - "*:7080": { - "application": name - } - }, - "applications": { - name: { - "type": "python", - "processes": { "spare": 0 }, - "path": "/app", - "module": "wsgi" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"*:7080": {"application": name}}, + "applications": { + name: { + "type": "python", + "processes": {"spare": 0}, + "path": "/app", + "module": "wsgi", + } + }, } - } - })) + ), + ) @unittest.expectedFailure def test_json_application_many(self): - self.skip_alerts.extend([ - r'eventfd.+failed', - r'epoll.+failed', - r'failed to apply' - ]) + self.skip_alerts.extend( + [r'eventfd.+failed', r'epoll.+failed', r'failed to apply'] + ) apps = 999 conf = { - "applications": - {"app-" + str(a): { + "applications": { + "app-" + + str(a): { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module": "wsgi" - } for a in range(apps) + "module": "wsgi", + } + for a in range(apps) }, "listeners": { - "*:" + str(7000 + a): { - "application": "app-" + str(a) - } for a in range(apps) - } + "*:" + str(7000 + a): {"application": "app-" + str(a)} + for a in range(apps) + }, } self.assertIn('success', self.conf(conf)) def test_json_application_many2(self): - self.skip_alerts.extend([ - r'eventfd.+failed', - r'epoll.+failed', - r'failed to apply' - ]) + self.skip_alerts.extend( + [r'eventfd.+failed', r'epoll.+failed', r'failed to apply'] + ) conf = { - "applications": - {"app-" + str(a): { + "applications": { + "app-" + + str(a): { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module": "wsgi" - } for a in range(999) - }, - "listeners": { - "*:7001": { - "application": "app-1" + "module": "wsgi", } - } + for a in range(999) + }, + "listeners": {"*:7001": {"application": "app-1"}}, } self.assertIn('success', self.conf(conf)) + if __name__ == '__main__': TestUnitConfiguration.main() diff --git a/test/test_go_application.py b/test/test_go_application.py index 1ecc2536..9b39b238 100644 --- a/test/test_go_application.py +++ b/test/test_go_application.py @@ -1,8 +1,8 @@ import unittest import unit -class TestUnitGoApplication(unit.TestUnitApplicationGo): +class TestUnitGoApplication(unit.TestUnitApplicationGo): def setUpClass(): unit.TestUnit().check_modules('go') @@ -11,12 +11,15 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] @@ -25,21 +28,28 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') - - self.assertDictEqual(headers, { - 'Content-Length': str(len(body)), - 'Content-Type': 'text/html', - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Server-Protocol-Major': '1', - 'Server-Protocol-Minor': '1', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, 'headers') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) + + self.assertDictEqual( + headers, + { + 'Content-Length': str(len(body)), + 'Content-Type': 'text/html', + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Server-Protocol-Major': '1', + 'Server-Protocol-Minor': '1', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_go_application_get_variables(self): @@ -53,11 +63,14 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): def test_go_application_post_variables(self): self.load('post_variables') - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'application/x-www-form-urlencoded', - 'Connection': 'close' - }, body='var1=val1&var2=&var3') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'application/x-www-form-urlencoded', + 'Connection': 'close', + }, + body='var1=val1&var2=&var3', + ) self.assertEqual(resp['headers']['X-Var-1'], 'val1', 'POST variables') self.assertEqual(resp['headers']['X-Var-2'], '', 'POST variables 2') @@ -69,36 +82,47 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): resp = self.get() self.assertEqual(resp['status'], 404, '404 status') - self.assertRegex(resp['body'], r'<title>404 Not Found</title>', - '404 body') + self.assertRegex( + resp['body'], r'<title>404 Not Found</title>', '404 body' + ) def test_go_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') def test_go_application_cookies(self): self.load('cookies') - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'var1=val1; var2=val2', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'var1=val1; var2=val2', + 'Connection': 'close', + } + ) self.assertEqual(resp['headers']['X-Cookie-1'], 'val1', 'cookie 1') self.assertEqual(resp['headers']['X-Cookie-2'], 'val2', 'cookie 2') @@ -106,15 +130,22 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): def test_go_application_command_line_arguments_type(self): self.load('command_line_arguments') - self.assertIn('error', self.conf(''"a b c", - 'applications/command_line_arguments/arguments'), 'arguments type') + self.assertIn( + 'error', + self.conf( + '' "a b c", 'applications/command_line_arguments/arguments' + ), + 'arguments type', + ) def test_go_application_command_line_arguments_0(self): self.load('command_line_arguments') - self.assertEqual(self.get()['headers']['X-Arg-0'], + self.assertEqual( + self.get()['headers']['X-Arg-0'], self.conf_get('applications/command_line_arguments/executable'), - 'argument 0') + 'argument 0', + ) def test_go_application_command_line_arguments(self): self.load('command_line_arguments') @@ -123,11 +154,14 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): arg2 = '--cc-opt=\'-O0 -DNXT_DEBUG_MEMORY=1 -fsanitize=address\'' arg3 = '--debug' - self.conf('["' + arg1 + '", "' + arg2 + '", "' + arg3 + '"]', - 'applications/command_line_arguments/arguments') + self.conf( + '["' + arg1 + '", "' + arg2 + '", "' + arg3 + '"]', + 'applications/command_line_arguments/arguments', + ) - self.assertEqual(self.get()['body'], arg1 + ',' + arg2 + ',' + arg3, - 'arguments') + self.assertEqual( + self.get()['body'], arg1 + ',' + arg2 + ',' + arg3, 'arguments' + ) def test_go_application_command_line_arguments_change(self): self.load('command_line_arguments') @@ -144,8 +178,10 @@ class TestUnitGoApplication(unit.TestUnitApplicationGo): self.conf('[]', args_path) - self.assertEqual(self.get()['headers']['Content-Length'], '0', - 'arguments empty') + self.assertEqual( + self.get()['headers']['Content-Length'], '0', 'arguments empty' + ) + if __name__ == '__main__': TestUnitGoApplication.main() diff --git a/test/test_http_header.py b/test/test_http_header.py index f2294371..002a8826 100644 --- a/test/test_http_header.py +++ b/test/test_http_header.py @@ -1,216 +1,302 @@ import unittest import unit -class TestUnitHTTPHeader(unit.TestUnitApplicationPython): +class TestUnitHTTPHeader(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') def test_http_header_value_leading_sp(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': ' ,', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': ' ,', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value leading sp status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value leading sp custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value leading sp custom header', + ) def test_http_header_value_leading_htab(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': '\t,', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': '\t,', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value leading htab status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value leading htab custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value leading htab custom header', + ) def test_http_header_value_trailing_sp(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': ', ', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': ', ', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value trailing sp status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value trailing sp custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value trailing sp custom header', + ) def test_http_header_value_trailing_htab(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': ',\t', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': ',\t', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value trailing htab status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value trailing htab custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value trailing htab custom header', + ) def test_http_header_value_both_sp(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': ' , ', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': ' , ', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value both sp status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value both sp custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value both sp custom header', + ) def test_http_header_value_both_htab(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': '\t,\t', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': '\t,\t', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value both htab status') - self.assertEqual(resp['headers']['Custom-Header'], ',', - 'value both htab custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + ',', + 'value both htab custom header', + ) def test_http_header_value_chars(self): self.load('custom_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header': '(),/:;<=>?@[\]{}\t !#$%&\'*+-.^_`|~', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header': '(),/:;<=>?@[\]{}\t !#$%&\'*+-.^_`|~', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 200, 'value chars status') - self.assertEqual(resp['headers']['Custom-Header'], - '(),/:;<=>?@[\]{}\t !#$%&\'*+-.^_`|~', 'value chars custom header') + self.assertEqual( + resp['headers']['Custom-Header'], + '(),/:;<=>?@[\]{}\t !#$%&\'*+-.^_`|~', + 'value chars custom header', + ) def test_http_header_value_chars_edge(self): self.load('custom_header') - resp = self.http(b"""GET / HTTP/1.1 + resp = self.http( + b"""GET / HTTP/1.1 Host: localhost Custom-Header: \x20\xFF Connection: close -""", raw=True, encoding='latin1') +""", + raw=True, + encoding='latin1', + ) self.assertEqual(resp['status'], 200, 'value chars edge status') - self.assertEqual(resp['headers']['Custom-Header'], '\xFF', - 'value chars edge') + self.assertEqual( + resp['headers']['Custom-Header'], '\xFF', 'value chars edge' + ) def test_http_header_value_chars_below(self): self.load('custom_header') - resp = self.http(b"""GET / HTTP/1.1 + resp = self.http( + b"""GET / HTTP/1.1 Host: localhost Custom-Header: \x1F Connection: close -""", raw=True) +""", + raw=True, + ) self.assertEqual(resp['status'], 400, 'value chars below') def test_http_header_field_leading_sp(self): self.load('empty') - resp = self.get(headers={ - 'Host': 'localhost', - ' Custom-Header': 'blah', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + ' Custom-Header': 'blah', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 400, 'field leading sp') def test_http_header_field_leading_htab(self): self.load('empty') - resp = self.get(headers={ - 'Host': 'localhost', - '\tCustom-Header': 'blah', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + '\tCustom-Header': 'blah', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 400, 'field leading htab') def test_http_header_field_trailing_sp(self): self.load('empty') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header ': 'blah', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header ': 'blah', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 400, 'field trailing sp') def test_http_header_field_trailing_htab(self): self.load('empty') - resp = self.get(headers={ - 'Host': 'localhost', - 'Custom-Header\t': 'blah', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Custom-Header\t': 'blah', + 'Connection': 'close', + } + ) self.assertEqual(resp['status'], 400, 'field trailing htab') def test_http_header_content_length_big(self): self.load('empty') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Length': str(2 ** 64), - 'Connection': 'close' - }, body='X' * 1000)['status'], 400, 'Content-Length big') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Length': str(2 ** 64), + 'Connection': 'close', + }, + body='X' * 1000, + )['status'], + 400, + 'Content-Length big', + ) def test_http_header_content_length_negative(self): self.load('empty') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Length': '-100', - 'Connection': 'close' - }, body='X' * 1000)['status'], 400, 'Content-Length negative') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Length': '-100', + 'Connection': 'close', + }, + body='X' * 1000, + )['status'], + 400, + 'Content-Length negative', + ) def test_http_header_content_length_text(self): self.load('empty') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Length': 'blah', - 'Connection': 'close' - }, body='X' * 1000)['status'], 400, 'Content-Length text') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Length': 'blah', + 'Connection': 'close', + }, + body='X' * 1000, + )['status'], + 400, + 'Content-Length text', + ) def test_http_header_content_length_multiple_values(self): self.load('empty') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Length': '41, 42', - 'Connection': 'close' - }, body='X' * 1000)['status'], 400, 'Content-Length multiple value') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Length': '41, 42', + 'Connection': 'close', + }, + body='X' * 1000, + )['status'], + 400, + 'Content-Length multiple value', + ) def test_http_header_content_length_multiple_fields(self): self.load('empty') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Length': ['41', '42'], - 'Connection': 'close' - }, body='X' * 1000)['status'], 400, 'Content-Length multiple fields') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Length': ['41', '42'], + 'Connection': 'close', + }, + body='X' * 1000, + )['status'], + 400, + 'Content-Length multiple fields', + ) def test_http_header_host_absent(self): self.load('host') @@ -218,146 +304,182 @@ Connection: close resp = self.get(headers={'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host absent status') - self.assertNotEqual(resp['headers']['X-Server-Name'], '', - 'Host absent SERVER_NAME') + self.assertNotEqual( + resp['headers']['X-Server-Name'], '', 'Host absent SERVER_NAME' + ) def test_http_header_host_empty(self): self.load('host') - resp = self.get(headers={ - 'Host': '', - 'Connection': 'close' - }) + resp = self.get(headers={'Host': '', 'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host empty status') - self.assertNotEqual(resp['headers']['X-Server-Name'], '', - 'Host empty SERVER_NAME') + self.assertNotEqual( + resp['headers']['X-Server-Name'], '', 'Host empty SERVER_NAME' + ) def test_http_header_host_big(self): self.load('empty') - self.assertEqual(self.get(headers={ - 'Host': 'X' * 10000, - 'Connection': 'close' - })['status'], 431, 'Host big') + self.assertEqual( + self.get(headers={'Host': 'X' * 10000, 'Connection': 'close'})[ + 'status' + ], + 431, + 'Host big', + ) def test_http_header_host_port(self): self.load('host') - resp = self.get(headers={ - 'Host': 'exmaple.com:7080', - 'Connection': 'close' - }) + resp = self.get( + headers={'Host': 'exmaple.com:7080', 'Connection': 'close'} + ) self.assertEqual(resp['status'], 200, 'Host port status') - self.assertEqual(resp['headers']['X-Server-Name'], 'exmaple.com', - 'Host port SERVER_NAME') - self.assertEqual(resp['headers']['X-Http-Host'], 'exmaple.com:7080', - 'Host port HTTP_HOST') + self.assertEqual( + resp['headers']['X-Server-Name'], + 'exmaple.com', + 'Host port SERVER_NAME', + ) + self.assertEqual( + resp['headers']['X-Http-Host'], + 'exmaple.com:7080', + 'Host port HTTP_HOST', + ) def test_http_header_host_port_empty(self): self.load('host') - resp = self.get(headers={ - 'Host': 'exmaple.com:', - 'Connection': 'close' - }) + resp = self.get( + headers={'Host': 'exmaple.com:', 'Connection': 'close'} + ) self.assertEqual(resp['status'], 200, 'Host port empty status') - self.assertEqual(resp['headers']['X-Server-Name'], 'exmaple.com', - 'Host port empty SERVER_NAME') - self.assertEqual(resp['headers']['X-Http-Host'], 'exmaple.com:', - 'Host port empty HTTP_HOST') + self.assertEqual( + resp['headers']['X-Server-Name'], + 'exmaple.com', + 'Host port empty SERVER_NAME', + ) + self.assertEqual( + resp['headers']['X-Http-Host'], + 'exmaple.com:', + 'Host port empty HTTP_HOST', + ) def test_http_header_host_literal(self): self.load('host') - resp = self.get(headers={ - 'Host': '127.0.0.1', - 'Connection': 'close' - }) + resp = self.get(headers={'Host': '127.0.0.1', 'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host literal status') - self.assertEqual(resp['headers']['X-Server-Name'], '127.0.0.1', - 'Host literal SERVER_NAME') + self.assertEqual( + resp['headers']['X-Server-Name'], + '127.0.0.1', + 'Host literal SERVER_NAME', + ) def test_http_header_host_literal_ipv6(self): self.load('host') - resp = self.get(headers={ - 'Host': '[::1]:7080', - 'Connection': 'close' - }) + resp = self.get(headers={'Host': '[::1]:7080', 'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host literal ipv6 status') - self.assertEqual(resp['headers']['X-Server-Name'], '[::1]', - 'Host literal ipv6 SERVER_NAME') - self.assertEqual(resp['headers']['X-Http-Host'], '[::1]:7080', - 'Host literal ipv6 HTTP_HOST') + self.assertEqual( + resp['headers']['X-Server-Name'], + '[::1]', + 'Host literal ipv6 SERVER_NAME', + ) + self.assertEqual( + resp['headers']['X-Http-Host'], + '[::1]:7080', + 'Host literal ipv6 HTTP_HOST', + ) def test_http_header_host_trailing_period(self): self.load('host') - resp = self.get(headers={ - 'Host': '127.0.0.1.', - 'Connection': 'close' - }) + resp = self.get(headers={'Host': '127.0.0.1.', 'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host trailing period status') - self.assertEqual(resp['headers']['X-Server-Name'], '127.0.0.1', - 'Host trailing period SERVER_NAME') - self.assertEqual(resp['headers']['X-Http-Host'], '127.0.0.1.', - 'Host trailing period HTTP_HOST') + self.assertEqual( + resp['headers']['X-Server-Name'], + '127.0.0.1', + 'Host trailing period SERVER_NAME', + ) + self.assertEqual( + resp['headers']['X-Http-Host'], + '127.0.0.1.', + 'Host trailing period HTTP_HOST', + ) def test_http_header_host_trailing_period_2(self): self.load('host') - resp = self.get(headers={ - 'Host': 'EXAMPLE.COM.', - 'Connection': 'close' - }) + resp = self.get( + headers={'Host': 'EXAMPLE.COM.', 'Connection': 'close'} + ) self.assertEqual(resp['status'], 200, 'Host trailing period 2 status') - self.assertEqual(resp['headers']['X-Server-Name'], 'example.com', - 'Host trailing period 2 SERVER_NAME') - self.assertEqual(resp['headers']['X-Http-Host'], 'EXAMPLE.COM.', - 'Host trailing period 2 HTTP_HOST') + self.assertEqual( + resp['headers']['X-Server-Name'], + 'example.com', + 'Host trailing period 2 SERVER_NAME', + ) + self.assertEqual( + resp['headers']['X-Http-Host'], + 'EXAMPLE.COM.', + 'Host trailing period 2 HTTP_HOST', + ) def test_http_header_host_case_insensitive(self): self.load('host') - resp = self.get(headers={ - 'Host': 'EXAMPLE.COM', - 'Connection': 'close' - }) + resp = self.get(headers={'Host': 'EXAMPLE.COM', 'Connection': 'close'}) self.assertEqual(resp['status'], 200, 'Host case insensitive') - self.assertEqual(resp['headers']['X-Server-Name'], 'example.com', - 'Host case insensitive SERVER_NAME') + self.assertEqual( + resp['headers']['X-Server-Name'], + 'example.com', + 'Host case insensitive SERVER_NAME', + ) def test_http_header_host_double_dot(self): self.load('empty') - self.assertEqual(self.get(headers={ - 'Host': '127.0.0..1', - 'Connection': 'close' - })['status'], 400, 'Host double dot') + self.assertEqual( + self.get(headers={'Host': '127.0.0..1', 'Connection': 'close'})[ + 'status' + ], + 400, + 'Host double dot', + ) def test_http_header_host_slash(self): self.load('empty') - self.assertEqual(self.get(headers={ - 'Host': '/localhost', - 'Connection': 'close' - })['status'], 400, 'Host slash') + self.assertEqual( + self.get(headers={'Host': '/localhost', 'Connection': 'close'})[ + 'status' + ], + 400, + 'Host slash', + ) def test_http_header_host_multiple_fields(self): self.load('empty') - self.assertEqual(self.get(headers={ - 'Host': ['localhost', 'example.com'], - 'Connection': 'close' - })['status'], 400, 'Host multiple fields') + self.assertEqual( + self.get( + headers={ + 'Host': ['localhost', 'example.com'], + 'Connection': 'close', + } + )['status'], + 400, + 'Host multiple fields', + ) + if __name__ == '__main__': TestUnitHTTPHeader.main() diff --git a/test/test_java_application.py b/test/test_java_application.py index d603ed0f..05e4aadd 100644 --- a/test/test_java_application.py +++ b/test/test_java_application.py @@ -2,19 +2,21 @@ import time import unittest import unit -class TestUnitJavaApplication(unit.TestUnitApplicationJava): +class TestUnitJavaApplication(unit.TestUnitApplicationJava): def setUpClass(): unit.TestUnit().check_modules('java') def test_java_application_cookies(self): self.load('cookies') - headers = self.get(headers={ - 'Cookie': 'var1=val1; var2=val2', - 'Host': 'localhost', - 'Connection': 'close' - })['headers'] + headers = self.get( + headers={ + 'Cookie': 'var1=val1; var2=val2', + 'Host': 'localhost', + 'Connection': 'close', + } + )['headers'] self.assertEqual(headers['X-Cookie-1'], 'val1', 'cookie 1') self.assertEqual(headers['X-Cookie-2'], 'val2', 'cookie 2') @@ -27,33 +29,46 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): self.assertEqual(headers['X-Filter-Before'], '1', 'filter before') self.assertEqual(headers['X-Filter-After'], '1', 'filter after') - self.assertEqual(self.get(url='/test')['headers']['X-Filter-After'], - '0', 'filter after 2') + self.assertEqual( + self.get(url='/test')['headers']['X-Filter-After'], + '0', + 'filter after 2', + ) def test_java_application_get_variables(self): self.load('get_params') - headers = self.get(url='/?var1=val1&var2=&var4=val4&var4=foo')['headers'] + headers = self.get(url='/?var1=val1&var2=&var4=val4&var4=foo')[ + 'headers' + ] self.assertEqual(headers['X-Var-1'], 'val1', 'GET variables') self.assertEqual(headers['X-Var-2'], 'true', 'GET variables 2') self.assertEqual(headers['X-Var-3'], 'false', 'GET variables 3') - self.assertEqual(headers['X-Param-Names'], 'var4 var2 var1 ', - 'getParameterNames') - self.assertEqual(headers['X-Param-Values'], 'val4 foo ', - 'getParameterValues') - self.assertEqual(headers['X-Param-Map'], - 'var2= var1=val1 var4=val4,foo ', 'getParameterMap') + self.assertEqual( + headers['X-Param-Names'], 'var4 var2 var1 ', 'getParameterNames' + ) + self.assertEqual( + headers['X-Param-Values'], 'val4 foo ', 'getParameterValues' + ) + self.assertEqual( + headers['X-Param-Map'], + 'var2= var1=val1 var4=val4,foo ', + 'getParameterMap', + ) def test_java_application_post_variables(self): self.load('post_params') - headers = self.post(headers={ - 'Content-Type': 'application/x-www-form-urlencoded', - 'Host': 'localhost', - 'Connection': 'close' - }, body='var1=val1&var2=')['headers'] + headers = self.post( + headers={ + 'Content-Type': 'application/x-www-form-urlencoded', + 'Host': 'localhost', + 'Connection': 'close', + }, + body='var1=val1&var2=', + )['headers'] self.assertEqual(headers['X-Var-1'], 'val1', 'POST variables') self.assertEqual(headers['X-Var-2'], 'true', 'POST variables 2') @@ -68,15 +83,20 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): self.assertEqual(headers['X-Var-1'], 'null', 'variable empty') self.assertEqual(headers['X-Session-New'], 'true', 'session create') - headers = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }, url='/?var1=val2')['headers'] + headers = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + }, + url='/?var1=val2', + )['headers'] self.assertEqual(headers['X-Var-1'], 'val1', 'variable') self.assertEqual(headers['X-Session-New'], 'false', 'session resume') - self.assertEqual(session_id, headers['X-Session-Id'], 'session same id') + self.assertEqual( + session_id, headers['X-Session-Id'], 'session same id' + ) def test_java_application_session_active(self): self.load('session_inactive') @@ -85,46 +105,63 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): session_id = resp['headers']['X-Session-Id'] self.assertEqual(resp['status'], 200, 'session init') - self.assertEqual(resp['headers']['X-Session-Interval'], '2', - 'session interval') - self.assertLess(abs(self.date_to_sec_epoch( - resp['headers']['X-Session-Last-Access-Time']) - self.sec_epoch()), - 5, 'session last access time') + self.assertEqual( + resp['headers']['X-Session-Interval'], '2', 'session interval' + ) + self.assertLess( + abs( + self.date_to_sec_epoch( + resp['headers']['X-Session-Last-Access-Time'] + ) + - self.sec_epoch() + ), + 5, + 'session last access time', + ) time.sleep(1) - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + } + ) - self.assertEqual(resp['headers']['X-Session-Id'], session_id, - 'session active') + self.assertEqual( + resp['headers']['X-Session-Id'], session_id, 'session active' + ) session_id = resp['headers']['X-Session-Id'] time.sleep(1) - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + } + ) - self.assertEqual(resp['headers']['X-Session-Id'], session_id, - 'session active 2') + self.assertEqual( + resp['headers']['X-Session-Id'], session_id, 'session active 2' + ) time.sleep(1) - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + } + ) - self.assertEqual(resp['headers']['X-Session-Id'], session_id, - 'session active 3') + self.assertEqual( + resp['headers']['X-Session-Id'], session_id, 'session active 3' + ) def test_java_application_session_inactive(self): self.load('session_inactive') @@ -134,14 +171,17 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): time.sleep(3) - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + } + ) - self.assertNotEqual(resp['headers']['X-Session-Id'], session_id, - 'session inactive') + self.assertNotEqual( + resp['headers']['X-Session-Id'], session_id, 'session inactive' + ) def test_java_application_session_invalidate(self): self.load('session_invalidate') @@ -149,14 +189,17 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): resp = self.get() session_id = resp['headers']['X-Session-Id'] - resp = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + } + ) - self.assertNotEqual(resp['headers']['X-Session-Id'], session_id, - 'session invalidate') + self.assertNotEqual( + resp['headers']['X-Session-Id'], session_id, 'session invalidate' + ) def test_java_application_session_listeners(self): self.load('session_listeners') @@ -164,30 +207,42 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/test?var1=val1')['headers'] session_id = headers['X-Session-Id'] - self.assertEqual(headers['X-Session-Created'], session_id, - 'session create') - self.assertEqual(headers['X-Attr-Added'], 'var1=val1', - 'attribute add') - - headers = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }, url='/?var1=val2')['headers'] - - self.assertEqual(session_id, headers['X-Session-Id'], 'session same id') - self.assertEqual(headers['X-Attr-Replaced'], 'var1=val1', - 'attribute replace') + self.assertEqual( + headers['X-Session-Created'], session_id, 'session create' + ) + self.assertEqual(headers['X-Attr-Added'], 'var1=val1', 'attribute add') + + headers = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + }, + url='/?var1=val2', + )['headers'] - headers = self.get(headers={ - 'Host': 'localhost', - 'Cookie': 'JSESSIONID=' + session_id, - 'Connection': 'close' - }, url='/')['headers'] + self.assertEqual( + session_id, headers['X-Session-Id'], 'session same id' + ) + self.assertEqual( + headers['X-Attr-Replaced'], 'var1=val1', 'attribute replace' + ) + + headers = self.get( + headers={ + 'Host': 'localhost', + 'Cookie': 'JSESSIONID=' + session_id, + 'Connection': 'close', + }, + url='/', + )['headers'] - self.assertEqual(session_id, headers['X-Session-Id'], 'session same id') - self.assertEqual(headers['X-Attr-Removed'], 'var1=val2', - 'attribute remove') + self.assertEqual( + session_id, headers['X-Session-Id'], 'session same id' + ) + self.assertEqual( + headers['X-Attr-Removed'], 'var1=val2', 'attribute remove' + ) def test_java_application_jsp(self): self.load('jsp') @@ -202,15 +257,23 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/foo/bar/index.html')['headers'] self.assertEqual(headers['X-Id'], 'servlet1', '#1 Servlet1 request') - self.assertEqual(headers['X-Request-URI'], '/foo/bar/index.html', '#1 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/foo/bar', '#1 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/foo/bar/index.html', '#1 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/foo/bar', '#1 servlet path' + ) self.assertEqual(headers['X-Path-Info'], '/index.html', '#1 path info') headers = self.get(url='/foo/bar/index.bop')['headers'] self.assertEqual(headers['X-Id'], 'servlet1', '#2 Servlet1 request') - self.assertEqual(headers['X-Request-URI'], '/foo/bar/index.bop', '#2 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/foo/bar', '#2 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/foo/bar/index.bop', '#2 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/foo/bar', '#2 servlet path' + ) self.assertEqual(headers['X-Path-Info'], '/index.bop', '#2 path info') headers = self.get(url='/baz')['headers'] @@ -223,42 +286,64 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/baz/index.html')['headers'] self.assertEqual(headers['X-Id'], 'servlet2', '#4 Servlet2 request') - self.assertEqual(headers['X-Request-URI'], '/baz/index.html', '#4 request URI') + self.assertEqual( + headers['X-Request-URI'], '/baz/index.html', '#4 request URI' + ) self.assertEqual(headers['X-Servlet-Path'], '/baz', '#4 servlet path') self.assertEqual(headers['X-Path-Info'], '/index.html', '#4 path info') headers = self.get(url='/catalog')['headers'] self.assertEqual(headers['X-Id'], 'servlet3', '#5 Servlet3 request') - self.assertEqual(headers['X-Request-URI'], '/catalog', '#5 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/catalog', '#5 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/catalog', '#5 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/catalog', '#5 servlet path' + ) self.assertEqual(headers['X-Path-Info'], 'null', '#5 path info') headers = self.get(url='/catalog/index.html')['headers'] self.assertEqual(headers['X-Id'], 'default', '#6 default request') - self.assertEqual(headers['X-Request-URI'], '/catalog/index.html', '#6 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/catalog/index.html', '#6 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/catalog/index.html', '#6 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/catalog/index.html', '#6 servlet path' + ) self.assertEqual(headers['X-Path-Info'], 'null', '#6 path info') headers = self.get(url='/catalog/racecar.bop')['headers'] self.assertEqual(headers['X-Id'], 'servlet4', '#7 servlet4 request') - self.assertEqual(headers['X-Request-URI'], '/catalog/racecar.bop', '#7 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/catalog/racecar.bop', '#7 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/catalog/racecar.bop', '#7 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], + '/catalog/racecar.bop', + '#7 servlet path', + ) self.assertEqual(headers['X-Path-Info'], 'null', '#7 path info') - headers = self.get( url='/index.bop')['headers'] + headers = self.get(url='/index.bop')['headers'] self.assertEqual(headers['X-Id'], 'servlet4', '#8 servlet4 request') - self.assertEqual(headers['X-Request-URI'], '/index.bop', '#8 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/index.bop', '#8 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/index.bop', '#8 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/index.bop', '#8 servlet path' + ) self.assertEqual(headers['X-Path-Info'], 'null', '#8 path info') headers = self.get(url='/foo/baz')['headers'] self.assertEqual(headers['X-Id'], 'servlet0', '#9 servlet0 request') - self.assertEqual(headers['X-Request-URI'], '/foo/baz', '#9 request URI') + self.assertEqual( + headers['X-Request-URI'], '/foo/baz', '#9 request URI' + ) self.assertEqual(headers['X-Servlet-Path'], '/foo', '#9 servlet path') self.assertEqual(headers['X-Path-Info'], '/baz', '#9 path info') @@ -272,8 +357,12 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/index.bop/')['headers'] self.assertEqual(headers['X-Id'], 'default', '#11 default request') - self.assertEqual(headers['X-Request-URI'], '/index.bop/', '#11 request URI') - self.assertEqual(headers['X-Servlet-Path'], '/index.bop/', '#11 servlet path') + self.assertEqual( + headers['X-Request-URI'], '/index.bop/', '#11 request URI' + ) + self.assertEqual( + headers['X-Servlet-Path'], '/index.bop/', '#11 servlet path' + ) self.assertEqual(headers['X-Path-Info'], 'null', '#11 path info') def test_java_application_header(self): @@ -281,10 +370,18 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get()['headers'] - self.assertEqual(headers['X-Set-Utf8-Value'], '????', 'set Utf8 header value') - self.assertEqual(headers['X-Set-Utf8-Name-???'], 'x', 'set Utf8 header name') - self.assertEqual(headers['X-Add-Utf8-Value'], '????', 'add Utf8 header value') - self.assertEqual(headers['X-Add-Utf8-Name-???'], 'y', 'add Utf8 header name') + self.assertEqual( + headers['X-Set-Utf8-Value'], '????', 'set Utf8 header value' + ) + self.assertEqual( + headers['X-Set-Utf8-Name-???'], 'x', 'set Utf8 header name' + ) + self.assertEqual( + headers['X-Add-Utf8-Value'], '????', 'add Utf8 header value' + ) + self.assertEqual( + headers['X-Add-Utf8-Name-???'], 'y', 'add Utf8 header name' + ) self.assertEqual(headers['X-Add-Test'], 'v1', 'add null header') self.assertEqual('X-Set-Test1' in headers, False, 'set null header') self.assertEqual(headers['X-Set-Test2'], '', 'set empty header') @@ -294,52 +391,135 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/1')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=utf-8', '#1 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=utf-8', '#1 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'utf-8', '#1 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=utf-8', + '#1 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=utf-8', + '#1 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], 'utf-8', '#1 response charset' + ) headers = self.get(url='/2')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=iso-8859-1', '#2 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=iso-8859-1', '#2 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'iso-8859-1', '#2 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=iso-8859-1', + '#2 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=iso-8859-1', + '#2 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], + 'iso-8859-1', + '#2 response charset', + ) headers = self.get(url='/3')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=windows-1251', '#3 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=windows-1251', '#3 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'windows-1251', '#3 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=windows-1251', + '#3 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=windows-1251', + '#3 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], + 'windows-1251', + '#3 response charset', + ) headers = self.get(url='/4')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=windows-1251', '#4 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=windows-1251', '#4 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'windows-1251', '#4 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=windows-1251', + '#4 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=windows-1251', + '#4 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], + 'windows-1251', + '#4 response charset', + ) headers = self.get(url='/5')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=iso-8859-1', '#5 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=iso-8859-1', '#5 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'iso-8859-1', '#5 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=iso-8859-1', + '#5 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=iso-8859-1', + '#5 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], + 'iso-8859-1', + '#5 response charset', + ) headers = self.get(url='/6')['headers'] - self.assertEqual('Content-Type' in headers, False, '#6 no Content-Type header') - self.assertEqual('X-Content-Type' in headers, False, '#6 no response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'utf-8', '#6 response charset') - + self.assertEqual( + 'Content-Type' in headers, False, '#6 no Content-Type header' + ) + self.assertEqual( + 'X-Content-Type' in headers, False, '#6 no response Content-Type' + ) + self.assertEqual( + headers['X-Character-Encoding'], 'utf-8', '#6 response charset' + ) headers = self.get(url='/7')['headers'] - self.assertEqual(headers['Content-Type'], 'text/plain;charset=utf-8', '#7 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/plain;charset=utf-8', '#7 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'utf-8', '#7 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/plain;charset=utf-8', + '#7 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/plain;charset=utf-8', + '#7 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], 'utf-8', '#7 response charset' + ) headers = self.get(url='/8')['headers'] - self.assertEqual(headers['Content-Type'], 'text/html;charset=utf-8', '#8 Content-Type header') - self.assertEqual(headers['X-Content-Type'], 'text/html;charset=utf-8', '#8 response Content-Type') - self.assertEqual(headers['X-Character-Encoding'], 'utf-8', '#8 response charset') + self.assertEqual( + headers['Content-Type'], + 'text/html;charset=utf-8', + '#8 Content-Type header', + ) + self.assertEqual( + headers['X-Content-Type'], + 'text/html;charset=utf-8', + '#8 response Content-Type', + ) + self.assertEqual( + headers['X-Character-Encoding'], 'utf-8', '#8 response charset' + ) def test_java_application_welcome_files(self): self.load('welcome_files') @@ -352,8 +532,12 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): resp = self.get(url='/dir1/') - self.assertEqual('This is index.txt.' in resp['body'], True, 'dir1 index body') - self.assertEqual(resp['headers']['X-TXT-Filter'], '1', 'TXT Filter header') + self.assertEqual( + 'This is index.txt.' in resp['body'], True, 'dir1 index body' + ) + self.assertEqual( + resp['headers']['X-TXT-Filter'], '1', 'TXT Filter header' + ) headers = self.get(url='/dir2/')['headers'] @@ -362,124 +546,216 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = self.get(url='/dir3/')['headers'] - self.assertEqual(headers['X-App-Servlet'], '1', 'URL pattern overrides welcome file') + self.assertEqual( + headers['X-App-Servlet'], '1', 'URL pattern overrides welcome file' + ) headers = self.get(url='/dir4/')['headers'] - self.assertEqual('X-App-Servlet' in headers, False, 'Static welcome file served first') + self.assertEqual( + 'X-App-Servlet' in headers, + False, + 'Static welcome file served first', + ) headers = self.get(url='/dir5/')['headers'] - self.assertEqual(headers['X-App-Servlet'], '1', 'Servlet for welcome file served when no static file found') + self.assertEqual( + headers['X-App-Servlet'], + '1', + 'Servlet for welcome file served when no static file found', + ) def test_java_application_request_listeners(self): self.load('request_listeners') headers = self.get(url='/test1')['headers'] - self.assertEqual(headers['X-Request-Initialized'], '/test1', - 'request initialized event') - self.assertEqual(headers['X-Request-Destroyed'], '', - 'request destroyed event') - self.assertEqual(headers['X-Attr-Added'], '', - 'attribute added event') - self.assertEqual(headers['X-Attr-Removed'], '', - 'attribute removed event') - self.assertEqual(headers['X-Attr-Replaced'], '', - 'attribute replaced event') + self.assertEqual( + headers['X-Request-Initialized'], + '/test1', + 'request initialized event', + ) + self.assertEqual( + headers['X-Request-Destroyed'], '', 'request destroyed event' + ) + self.assertEqual(headers['X-Attr-Added'], '', 'attribute added event') + self.assertEqual( + headers['X-Attr-Removed'], '', 'attribute removed event' + ) + self.assertEqual( + headers['X-Attr-Replaced'], '', 'attribute replaced event' + ) headers = self.get(url='/test2?var1=1')['headers'] - self.assertEqual(headers['X-Request-Initialized'], '/test2', - 'request initialized event') - self.assertEqual(headers['X-Request-Destroyed'], '/test1', - 'request destroyed event') - self.assertEqual(headers['X-Attr-Added'], 'var=1;', - 'attribute added event') - self.assertEqual(headers['X-Attr-Removed'], 'var=1;', - 'attribute removed event') - self.assertEqual(headers['X-Attr-Replaced'], '', - 'attribute replaced event') + self.assertEqual( + headers['X-Request-Initialized'], + '/test2', + 'request initialized event', + ) + self.assertEqual( + headers['X-Request-Destroyed'], '/test1', 'request destroyed event' + ) + self.assertEqual( + headers['X-Attr-Added'], 'var=1;', 'attribute added event' + ) + self.assertEqual( + headers['X-Attr-Removed'], 'var=1;', 'attribute removed event' + ) + self.assertEqual( + headers['X-Attr-Replaced'], '', 'attribute replaced event' + ) headers = self.get(url='/test3?var1=1&var2=2')['headers'] - self.assertEqual(headers['X-Request-Initialized'], '/test3', - 'request initialized event') - self.assertEqual(headers['X-Request-Destroyed'], '/test2', - 'request destroyed event') - self.assertEqual(headers['X-Attr-Added'], 'var=1;', - 'attribute added event') - self.assertEqual(headers['X-Attr-Removed'], 'var=2;', - 'attribute removed event') - self.assertEqual(headers['X-Attr-Replaced'], 'var=1;', - 'attribute replaced event') + self.assertEqual( + headers['X-Request-Initialized'], + '/test3', + 'request initialized event', + ) + self.assertEqual( + headers['X-Request-Destroyed'], '/test2', 'request destroyed event' + ) + self.assertEqual( + headers['X-Attr-Added'], 'var=1;', 'attribute added event' + ) + self.assertEqual( + headers['X-Attr-Removed'], 'var=2;', 'attribute removed event' + ) + self.assertEqual( + headers['X-Attr-Replaced'], 'var=1;', 'attribute replaced event' + ) headers = self.get(url='/test4?var1=1&var2=2&var3=3')['headers'] - self.assertEqual(headers['X-Request-Initialized'], '/test4', - 'request initialized event') - self.assertEqual(headers['X-Request-Destroyed'], '/test3', - 'request destroyed event') - self.assertEqual(headers['X-Attr-Added'], 'var=1;', - 'attribute added event') - self.assertEqual(headers['X-Attr-Removed'], '', - 'attribute removed event') - self.assertEqual(headers['X-Attr-Replaced'], 'var=1;var=2;', - 'attribute replaced event') + self.assertEqual( + headers['X-Request-Initialized'], + '/test4', + 'request initialized event', + ) + self.assertEqual( + headers['X-Request-Destroyed'], '/test3', 'request destroyed event' + ) + self.assertEqual( + headers['X-Attr-Added'], 'var=1;', 'attribute added event' + ) + self.assertEqual( + headers['X-Attr-Removed'], '', 'attribute removed event' + ) + self.assertEqual( + headers['X-Attr-Replaced'], + 'var=1;var=2;', + 'attribute replaced event', + ) def test_java_application_request_uri_forward(self): self.load('forward') - resp = self.get(url='/fwd?uri=%2Fdata%2Ftest%3Furi%3Dnew_uri%26a%3D2%26b%3D3&a=1&c=4') + resp = self.get( + url='/fwd?uri=%2Fdata%2Ftest%3Furi%3Dnew_uri%26a%3D2%26b%3D3&a=1&c=4' + ) headers = resp['headers'] - self.assertEqual(headers['X-REQUEST-Id'], 'fwd', - 'initial request servlet mapping') - self.assertEqual(headers['X-Forward-To'], '/data/test?uri=new_uri&a=2&b=3', - 'forwarding triggered') - self.assertEqual(headers['X-REQUEST-Param-uri'], '/data/test?uri=new_uri&a=2&b=3', - 'original uri parameter') - self.assertEqual(headers['X-REQUEST-Param-a'], '1', - 'original a parameter') - self.assertEqual(headers['X-REQUEST-Param-c'], '4', - 'original c parameter') - - self.assertEqual(headers['X-FORWARD-Id'], 'data', - 'forward request servlet mapping') - self.assertEqual(headers['X-FORWARD-Request-URI'], '/data/test', - 'forward request uri') - self.assertEqual(headers['X-FORWARD-Servlet-Path'], '/data', - 'forward request servlet path') - self.assertEqual(headers['X-FORWARD-Path-Info'], '/test', - 'forward request path info') - self.assertEqual(headers['X-FORWARD-Query-String'], 'uri=new_uri&a=2&b=3', - 'forward request query string') - self.assertEqual(headers['X-FORWARD-Param-uri'], 'new_uri,/data/test?uri=new_uri&a=2&b=3', - 'forward uri parameter') - self.assertEqual(headers['X-FORWARD-Param-a'], '2,1', - 'forward a parameter') - self.assertEqual(headers['X-FORWARD-Param-b'], '3', - 'forward b parameter') - self.assertEqual(headers['X-FORWARD-Param-c'], '4', - 'forward c parameter') - - self.assertEqual(headers['X-javax.servlet.forward.request_uri'], '/fwd', - 'original request uri') - self.assertEqual(headers['X-javax.servlet.forward.context_path'], '', - 'original request context path') - self.assertEqual(headers['X-javax.servlet.forward.servlet_path'], '/fwd', - 'original request servlet path') - self.assertEqual(headers['X-javax.servlet.forward.path_info'], 'null', - 'original request path info') - self.assertEqual(headers['X-javax.servlet.forward.query_string'], 'uri=%2Fdata%2Ftest%3Furi%3Dnew_uri%26a%3D2%26b%3D3&a=1&c=4', - 'original request query') - - self.assertEqual('Before forwarding' in resp['body'], False, - 'discarded data added before forward() call') - self.assertEqual('X-After-Forwarding' in headers, False, - 'cannot add headers after forward() call') - self.assertEqual('After forwarding' in resp['body'], False, - 'cannot add data after forward() call') + self.assertEqual( + headers['X-REQUEST-Id'], 'fwd', 'initial request servlet mapping' + ) + self.assertEqual( + headers['X-Forward-To'], + '/data/test?uri=new_uri&a=2&b=3', + 'forwarding triggered', + ) + self.assertEqual( + headers['X-REQUEST-Param-uri'], + '/data/test?uri=new_uri&a=2&b=3', + 'original uri parameter', + ) + self.assertEqual( + headers['X-REQUEST-Param-a'], '1', 'original a parameter' + ) + self.assertEqual( + headers['X-REQUEST-Param-c'], '4', 'original c parameter' + ) + + self.assertEqual( + headers['X-FORWARD-Id'], 'data', 'forward request servlet mapping' + ) + self.assertEqual( + headers['X-FORWARD-Request-URI'], + '/data/test', + 'forward request uri', + ) + self.assertEqual( + headers['X-FORWARD-Servlet-Path'], + '/data', + 'forward request servlet path', + ) + self.assertEqual( + headers['X-FORWARD-Path-Info'], + '/test', + 'forward request path info', + ) + self.assertEqual( + headers['X-FORWARD-Query-String'], + 'uri=new_uri&a=2&b=3', + 'forward request query string', + ) + self.assertEqual( + headers['X-FORWARD-Param-uri'], + 'new_uri,/data/test?uri=new_uri&a=2&b=3', + 'forward uri parameter', + ) + self.assertEqual( + headers['X-FORWARD-Param-a'], '2,1', 'forward a parameter' + ) + self.assertEqual( + headers['X-FORWARD-Param-b'], '3', 'forward b parameter' + ) + self.assertEqual( + headers['X-FORWARD-Param-c'], '4', 'forward c parameter' + ) + + self.assertEqual( + headers['X-javax.servlet.forward.request_uri'], + '/fwd', + 'original request uri', + ) + self.assertEqual( + headers['X-javax.servlet.forward.context_path'], + '', + 'original request context path', + ) + self.assertEqual( + headers['X-javax.servlet.forward.servlet_path'], + '/fwd', + 'original request servlet path', + ) + self.assertEqual( + headers['X-javax.servlet.forward.path_info'], + 'null', + 'original request path info', + ) + self.assertEqual( + headers['X-javax.servlet.forward.query_string'], + 'uri=%2Fdata%2Ftest%3Furi%3Dnew_uri%26a%3D2%26b%3D3&a=1&c=4', + 'original request query', + ) + + self.assertEqual( + 'Before forwarding' in resp['body'], + False, + 'discarded data added before forward() call', + ) + self.assertEqual( + 'X-After-Forwarding' in headers, + False, + 'cannot add headers after forward() call', + ) + self.assertEqual( + 'After forwarding' in resp['body'], + False, + 'cannot add data after forward() call', + ) def test_java_application_named_dispatcher_forward(self): self.load('forward') @@ -487,39 +763,74 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): resp = self.get(url='/fwd?disp=name&uri=data') headers = resp['headers'] - self.assertEqual(headers['X-REQUEST-Id'], 'fwd', - 'initial request servlet mapping') - self.assertEqual(headers['X-Forward-To'], 'data', - 'forwarding triggered') - - self.assertEqual(headers['X-FORWARD-Id'], 'data', - 'forward request servlet mapping') - self.assertEqual(headers['X-FORWARD-Request-URI'], '/fwd', - 'forward request uri') - self.assertEqual(headers['X-FORWARD-Servlet-Path'], '/fwd', - 'forward request servlet path') - self.assertEqual(headers['X-FORWARD-Path-Info'], 'null', - 'forward request path info') - self.assertEqual(headers['X-FORWARD-Query-String'], 'disp=name&uri=data', - 'forward request query string') - - self.assertEqual(headers['X-javax.servlet.forward.request_uri'], 'null', - 'original request uri') - self.assertEqual(headers['X-javax.servlet.forward.context_path'], 'null', - 'original request context path') - self.assertEqual(headers['X-javax.servlet.forward.servlet_path'], 'null', - 'original request servlet path') - self.assertEqual(headers['X-javax.servlet.forward.path_info'], 'null', - 'original request path info') - self.assertEqual(headers['X-javax.servlet.forward.query_string'], 'null', - 'original request query') - - self.assertEqual('Before forwarding' in resp['body'], False, - 'discarded data added before forward() call') - self.assertEqual('X-After-Forwarding' in headers, False, - 'cannot add headers after forward() call') - self.assertEqual('After forwarding' in resp['body'], False, - 'cannot add data after forward() call') + self.assertEqual( + headers['X-REQUEST-Id'], 'fwd', 'initial request servlet mapping' + ) + self.assertEqual( + headers['X-Forward-To'], 'data', 'forwarding triggered' + ) + + self.assertEqual( + headers['X-FORWARD-Id'], 'data', 'forward request servlet mapping' + ) + self.assertEqual( + headers['X-FORWARD-Request-URI'], '/fwd', 'forward request uri' + ) + self.assertEqual( + headers['X-FORWARD-Servlet-Path'], + '/fwd', + 'forward request servlet path', + ) + self.assertEqual( + headers['X-FORWARD-Path-Info'], 'null', 'forward request path info' + ) + self.assertEqual( + headers['X-FORWARD-Query-String'], + 'disp=name&uri=data', + 'forward request query string', + ) + + self.assertEqual( + headers['X-javax.servlet.forward.request_uri'], + 'null', + 'original request uri', + ) + self.assertEqual( + headers['X-javax.servlet.forward.context_path'], + 'null', + 'original request context path', + ) + self.assertEqual( + headers['X-javax.servlet.forward.servlet_path'], + 'null', + 'original request servlet path', + ) + self.assertEqual( + headers['X-javax.servlet.forward.path_info'], + 'null', + 'original request path info', + ) + self.assertEqual( + headers['X-javax.servlet.forward.query_string'], + 'null', + 'original request query', + ) + + self.assertEqual( + 'Before forwarding' in resp['body'], + False, + 'discarded data added before forward() call', + ) + self.assertEqual( + 'X-After-Forwarding' in headers, + False, + 'cannot add headers after forward() call', + ) + self.assertEqual( + 'After forwarding' in resp['body'], + False, + 'cannot add data after forward() call', + ) def test_java_application_request_uri_include(self): self.load('include') @@ -528,31 +839,55 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = resp['headers'] body = resp['body'] - self.assertEqual(headers['X-REQUEST-Id'], 'inc', - 'initial request servlet mapping') - self.assertEqual(headers['X-Include'], '/data/test', - 'including triggered') - - self.assertEqual('X-INCLUDE-Id' in headers, False, - 'unable to add headers in include request') - - self.assertEqual('javax.servlet.include.request_uri: /data/test' in body, - True, 'include request uri') -# self.assertEqual('javax.servlet.include.context_path: ' in body, -# 'include request context path') - self.assertEqual('javax.servlet.include.servlet_path: /data' in body, - True, 'include request servlet path') - self.assertEqual('javax.servlet.include.path_info: /test' in body, - True, 'include request path info') - self.assertEqual('javax.servlet.include.query_string: null' in body, - True, 'include request query') - - self.assertEqual('Before include' in body, True, - 'preserve data added before include() call') - self.assertEqual(headers['X-After-Include'], 'you-should-see-this', - 'add headers after include() call') - self.assertEqual('After include' in body, True, - 'add data after include() call') + self.assertEqual( + headers['X-REQUEST-Id'], 'inc', 'initial request servlet mapping' + ) + self.assertEqual( + headers['X-Include'], '/data/test', 'including triggered' + ) + + self.assertEqual( + 'X-INCLUDE-Id' in headers, + False, + 'unable to add headers in include request', + ) + + self.assertEqual( + 'javax.servlet.include.request_uri: /data/test' in body, + True, + 'include request uri', + ) + # self.assertEqual('javax.servlet.include.context_path: ' in body, + # 'include request context path') + self.assertEqual( + 'javax.servlet.include.servlet_path: /data' in body, + True, + 'include request servlet path', + ) + self.assertEqual( + 'javax.servlet.include.path_info: /test' in body, + True, + 'include request path info', + ) + self.assertEqual( + 'javax.servlet.include.query_string: null' in body, + True, + 'include request query', + ) + + self.assertEqual( + 'Before include' in body, + True, + 'preserve data added before include() call', + ) + self.assertEqual( + headers['X-After-Include'], + 'you-should-see-this', + 'add headers after include() call', + ) + self.assertEqual( + 'After include' in body, True, 'add data after include() call' + ) def test_java_application_named_dispatcher_include(self): self.load('include') @@ -561,82 +896,134 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): headers = resp['headers'] body = resp['body'] - self.assertEqual(headers['X-REQUEST-Id'], 'inc', - 'initial request servlet mapping') - self.assertEqual(headers['X-Include'], 'data', - 'including triggered') - - self.assertEqual('X-INCLUDE-Id' in headers, False, - 'unable to add headers in include request') - - self.assertEqual('javax.servlet.include.request_uri: null' in body, - True, 'include request uri') -# self.assertEqual('javax.servlet.include.context_path: null' in body, -# 'include request context path') - self.assertEqual('javax.servlet.include.servlet_path: null' in body, - True, 'include request servlet path') - self.assertEqual('javax.servlet.include.path_info: null' in body, - True, 'include request path info') - self.assertEqual('javax.servlet.include.query_string: null' in body, - True, 'include request query') - - self.assertEqual('Before include' in body, True, - 'preserve data added before include() call') - self.assertEqual(headers['X-After-Include'], 'you-should-see-this', - 'add headers after include() call') - self.assertEqual('After include' in body, True, - 'add data after include() call') + self.assertEqual( + headers['X-REQUEST-Id'], 'inc', 'initial request servlet mapping' + ) + self.assertEqual(headers['X-Include'], 'data', 'including triggered') + + self.assertEqual( + 'X-INCLUDE-Id' in headers, + False, + 'unable to add headers in include request', + ) + + self.assertEqual( + 'javax.servlet.include.request_uri: null' in body, + True, + 'include request uri', + ) + # self.assertEqual('javax.servlet.include.context_path: null' in body, + # 'include request context path') + self.assertEqual( + 'javax.servlet.include.servlet_path: null' in body, + True, + 'include request servlet path', + ) + self.assertEqual( + 'javax.servlet.include.path_info: null' in body, + True, + 'include request path info', + ) + self.assertEqual( + 'javax.servlet.include.query_string: null' in body, + True, + 'include request query', + ) + + self.assertEqual( + 'Before include' in body, + True, + 'preserve data added before include() call', + ) + self.assertEqual( + headers['X-After-Include'], + 'you-should-see-this', + 'add headers after include() call', + ) + self.assertEqual( + 'After include' in body, True, 'add data after include() call' + ) def test_java_application_path_translation(self): self.load('path_translation') headers = self.get(url='/pt/test?path=/')['headers'] - self.assertEqual(headers['X-Servlet-Path'], '/pt', - 'matched servlet path') - self.assertEqual(headers['X-Path-Info'], '/test', - 'the rest of the path') - self.assertEqual(headers['X-Path-Translated'], + self.assertEqual( + headers['X-Servlet-Path'], '/pt', 'matched servlet path' + ) + self.assertEqual( + headers['X-Path-Info'], '/test', 'the rest of the path' + ) + self.assertEqual( + headers['X-Path-Translated'], headers['X-Real-Path'] + headers['X-Path-Info'], - 'translated path is the app root + path info') + 'translated path is the app root + path info', + ) self.assertEqual( headers['X-Resource-Paths'].endswith('/WEB-INF/, /index.html]'), - True, 'app root directory content') - self.assertEqual(headers['X-Resource-As-Stream'], 'null', - 'no resource stream for root path') + True, + 'app root directory content', + ) + self.assertEqual( + headers['X-Resource-As-Stream'], + 'null', + 'no resource stream for root path', + ) headers = self.get(url='/test?path=/none')['headers'] - self.assertEqual(headers['X-Servlet-Path'], '/test', - 'matched whole path') - self.assertEqual(headers['X-Path-Info'], 'null', - 'the rest of the path is null, whole path matched') - self.assertEqual(headers['X-Path-Translated'], 'null', - 'translated path is null because path info is null') - self.assertEqual(headers['X-Real-Path'].endswith('/none'), True, - 'read path is not null') - self.assertEqual(headers['X-Resource-Paths'], 'null', - 'no resource found') - self.assertEqual(headers['X-Resource-As-Stream'], 'null', - 'no resource stream') + self.assertEqual( + headers['X-Servlet-Path'], '/test', 'matched whole path' + ) + self.assertEqual( + headers['X-Path-Info'], + 'null', + 'the rest of the path is null, whole path matched', + ) + self.assertEqual( + headers['X-Path-Translated'], + 'null', + 'translated path is null because path info is null', + ) + self.assertEqual( + headers['X-Real-Path'].endswith('/none'), + True, + 'read path is not null', + ) + self.assertEqual( + headers['X-Resource-Paths'], 'null', 'no resource found' + ) + self.assertEqual( + headers['X-Resource-As-Stream'], 'null', 'no resource stream' + ) def test_java_application_query_string(self): self.load('query_string') - self.assertEqual(self.get(url='/?a=b')['headers']['X-Query-String'], - 'a=b', 'query string') + self.assertEqual( + self.get(url='/?a=b')['headers']['X-Query-String'], + 'a=b', + 'query string', + ) def test_java_application_query_empty(self): self.load('query_string') - self.assertEqual(self.get(url='/?')['headers']['X-Query-String'], '', - 'query string empty') + self.assertEqual( + self.get(url='/?')['headers']['X-Query-String'], + '', + 'query string empty', + ) def test_java_application_query_absent(self): self.load('query_string') - self.assertEqual(self.get()['headers']['X-Query-String'], 'null', - 'query string absent') + self.assertEqual( + self.get()['headers']['X-Query-String'], + 'null', + 'query string absent', + ) def test_java_application_empty(self): self.load('empty') @@ -646,19 +1033,27 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): def test_java_application_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Connection': 'keep-alive', - 'Content-Type': 'text/html', - 'Host': 'localhost' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + 'Host': 'localhost', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Connection': 'close', - 'Content-Type': 'text/html', - 'Host': 'localhost' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Connection': 'close', + 'Content-Type': 'text/html', + 'Host': 'localhost', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') @@ -675,11 +1070,17 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): def test_java_application_get_header(self): self.load('get_header') - self.assertEqual(self.get(headers={ - 'X-Header': 'blah', - 'Content-Type': 'text/html', - 'Host': 'localhost' - })['headers']['X-Reply'], 'blah', 'get header') + self.assertEqual( + self.get( + headers={ + 'X-Header': 'blah', + 'Content-Type': 'text/html', + 'Host': 'localhost', + } + )['headers']['X-Reply'], + 'blah', + 'get header', + ) def test_java_application_get_header_empty(self): self.load('get_header') @@ -689,11 +1090,13 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): def test_java_application_get_headers(self): self.load('get_headers') - headers = self.get(headers={ - 'X-Header': ['blah', 'blah'], - 'Content-Type': 'text/html', - 'Host': 'localhost' - })['headers'] + headers = self.get( + headers={ + 'X-Header': ['blah', 'blah'], + 'Content-Type': 'text/html', + 'Host': 'localhost', + } + )['headers'] self.assertEqual(headers['X-Reply-0'], 'blah', 'get headers') self.assertEqual(headers['X-Reply-1'], 'blah', 'get headers 2') @@ -701,35 +1104,46 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): def test_java_application_get_headers_empty(self): self.load('get_headers') - self.assertNotIn('X-Reply-0', self.get()['headers'], - 'get headers empty') + self.assertNotIn( + 'X-Reply-0', self.get()['headers'], 'get headers empty' + ) def test_java_application_get_header_names(self): self.load('get_header_names') headers = self.get()['headers'] - self.assertRegex(headers['X-Reply-0'], r'(?:Host|Connection)', - 'get header names') - self.assertRegex(headers['X-Reply-1'], r'(?:Host|Connection)', - 'get header names 2') - self.assertNotEqual(headers['X-Reply-0'], headers['X-Reply-1'], - 'get header names not equal') + self.assertRegex( + headers['X-Reply-0'], r'(?:Host|Connection)', 'get header names' + ) + self.assertRegex( + headers['X-Reply-1'], r'(?:Host|Connection)', 'get header names 2' + ) + self.assertNotEqual( + headers['X-Reply-0'], + headers['X-Reply-1'], + 'get header names not equal', + ) def test_java_application_get_header_names_empty(self): self.load('get_header_names') - self.assertNotIn('X-Reply-0', self.get(headers={})['headers'], - 'get header names empty') + self.assertNotIn( + 'X-Reply-0', + self.get(headers={})['headers'], + 'get header names empty', + ) def test_java_application_header_int(self): self.load('header_int') - headers = self.get(headers={ - 'X-Header': '2', - 'Content-Type': 'text/html', - 'Host': 'localhost' - })['headers'] + headers = self.get( + headers={ + 'X-Header': '2', + 'Content-Type': 'text/html', + 'Host': 'localhost', + } + )['headers'] self.assertEqual(headers['X-Set-Int'], '1', 'set int header') self.assertEqual(headers['X-Get-Int'], '2', 'get int header') @@ -739,15 +1153,21 @@ class TestUnitJavaApplication(unit.TestUnitApplicationJava): date = 'Fri, 15 Mar 2019 14:45:34 GMT' - headers = self.get(headers={ - 'X-Header': date, - 'Content-Type': 'text/html', - 'Host': 'localhost' - })['headers'] + headers = self.get( + headers={ + 'X-Header': date, + 'Content-Type': 'text/html', + 'Host': 'localhost', + } + )['headers'] - self.assertEqual(headers['X-Set-Date'], 'Thu, 01 Jan 1970 00:00:01 GMT', - 'set date header') + self.assertEqual( + headers['X-Set-Date'], + 'Thu, 01 Jan 1970 00:00:01 GMT', + 'set date header', + ) self.assertEqual(headers['X-Get-Date'], date, 'get date header') + if __name__ == '__main__': TestUnitJavaApplication.main() diff --git a/test/test_node_application.py b/test/test_node_application.py index cd64fefa..1acf374f 100644 --- a/test/test_node_application.py +++ b/test/test_node_application.py @@ -1,8 +1,8 @@ import unittest import unit -class TestUnitNodeApplication(unit.TestUnitApplicationNode): +class TestUnitNodeApplication(unit.TestUnitApplicationNode): def setUpClass(): u = unit.TestUnit().check_modules('node') @@ -10,8 +10,9 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): self.load('basic') resp = self.get() - self.assertEqual(resp['headers']['Content-Type'], 'text/plain', - 'basic header') + self.assertEqual( + resp['headers']['Content-Type'], 'text/plain', 'basic header' + ) self.assertEqual(resp['body'], 'Hello World\n', 'basic body') def test_node_application_seq(self): @@ -25,12 +26,15 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] @@ -39,24 +43,35 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) raw_headers = headers.pop('Request-Raw-Headers') - self.assertRegex(raw_headers, r'^(?:Host|localhost|Content-Type|' \ - 'text\/html|Custom-Header|blah|Content-Length|17|Connection|' \ - 'close|,)+$', 'raw headers') - - self.assertDictEqual(headers, { - 'Connection': 'close', - 'Content-Length': str(len(body)), - 'Content-Type': 'text/html', - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Custom-Header': 'blah' - }, 'headers') + self.assertRegex( + raw_headers, + r'^(?:Host|localhost|Content-Type|' + 'text\/html|Custom-Header|blah|Content-Length|17|Connection|' + 'close|,)+$', + 'raw headers', + ) + + self.assertDictEqual( + headers, + { + 'Connection': 'close', + 'Content-Length': str(len(body)), + 'Content-Type': 'text/html', + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Custom-Header': 'blah', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_node_application_get_variables(self): @@ -70,11 +85,14 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): def test_node_application_post_variables(self): self.load('post_variables') - resp = self.post(headers={ - 'Content-Type': 'application/x-www-form-urlencoded', - 'Host': 'localhost', - 'Connection': 'close' - }, body='var1=val1&var2=&var3') + resp = self.post( + headers={ + 'Content-Type': 'application/x-www-form-urlencoded', + 'Host': 'localhost', + 'Connection': 'close', + }, + body='var1=val1&var2=&var3', + ) self.assertEqual(resp['headers']['X-Var-1'], 'val1', 'POST variables') self.assertEqual(resp['headers']['X-Var-2'], '', 'POST variables 2') @@ -86,41 +104,56 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): resp = self.get() self.assertEqual(resp['status'], 404, '404 status') - self.assertRegex(resp['body'], r'<title>404 Not Found</title>', - '404 body') + self.assertRegex( + resp['body'], r'<title>404 Not Found</title>', '404 body' + ) def test_node_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') def test_node_application_write_buffer(self): self.load('write_buffer') - self.assertEqual(self.get()['body'], '6\r\nbuffer\r\n0\r\n\r\n', - 'write buffer') + self.assertEqual( + self.get()['body'], '6\r\nbuffer\r\n0\r\n\r\n', 'write buffer' + ) def test_node_application_write_callback(self): self.load('write_callback') - self.assertEqual(self.get()['body'], - '5\r\nhello\r\n5\r\nworld\r\n0\r\n\r\n', 'write callback order') - self.assertTrue(self.waitforfiles(self.testdir + '/node/callback'), - 'write callback') + self.assertEqual( + self.get()['body'], + '5\r\nhello\r\n5\r\nworld\r\n0\r\n\r\n', + 'write callback order', + ) + self.assertTrue( + self.waitforfiles(self.testdir + '/node/callback'), + 'write callback', + ) def test_node_application_write_before_write_head(self): self.load('write_before_write_head') @@ -136,17 +169,22 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): def test_node_application_write_return(self): self.load('write_return') - self.assertEqual(self.get()['body'], - '4\r\nbody\r\n4\r\ntrue\r\n0\r\n\r\n', 'write return') + self.assertEqual( + self.get()['body'], + '4\r\nbody\r\n4\r\ntrue\r\n0\r\n\r\n', + 'write return', + ) def test_node_application_remove_header(self): self.load('remove_header') - resp = self.get(headers={ - 'Host': 'localhost', - 'X-Remove': 'X-Header', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Host': 'localhost', + 'X-Remove': 'X-Header', + 'Connection': 'close', + } + ) self.assertEqual(resp['headers']['Was-Header'], 'true', 'was header') self.assertEqual(resp['headers']['Has-Header'], 'false', 'has header') self.assertFalse('X-Header' in resp['headers'], 'remove header') @@ -154,35 +192,48 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): def test_node_application_remove_header_nonexisting(self): self.load('remove_header') - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Remove': 'blah', - 'Connection': 'close' - })['headers']['Has-Header'], 'true', 'remove header nonexisting') + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Remove': 'blah', + 'Connection': 'close', + } + )['headers']['Has-Header'], + 'true', + 'remove header nonexisting', + ) def test_node_application_update_header(self): self.load('update_header') - self.assertEqual(self.get()['headers']['X-Header'], 'new', - 'update header') + self.assertEqual( + self.get()['headers']['X-Header'], 'new', 'update header' + ) def test_node_application_set_header_array(self): self.load('set_header_array') - self.assertListEqual(self.get()['headers']['Set-Cookie'], - ['tc=one,two,three', 'tc=four,five,six'], 'set header array') + self.assertListEqual( + self.get()['headers']['Set-Cookie'], + ['tc=one,two,three', 'tc=four,five,six'], + 'set header array', + ) @unittest.expectedFailure def test_node_application_status_message(self): self.load('status_message') - self.assertRegex(self.get(raw_resp=True), r'200 blah', 'status message') + self.assertRegex( + self.get(raw_resp=True), r'200 blah', 'status message' + ) def test_node_application_get_header_type(self): self.load('get_header_type') - self.assertEqual(self.get()['headers']['X-Type'], 'number', - 'get header type') + self.assertEqual( + self.get()['headers']['X-Type'], 'number', 'get header type' + ) def test_node_application_header_name_case(self): self.load('header_name_case') @@ -196,56 +247,89 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): def test_node_application_promise_handler(self): self.load('promise_handler') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, body='callback')['status'], 200, 'promise handler request') - self.assertTrue(self.waitforfiles(self.testdir + '/node/callback'), - 'promise handler') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + body='callback', + )['status'], + 200, + 'promise handler request', + ) + self.assertTrue( + self.waitforfiles(self.testdir + '/node/callback'), + 'promise handler', + ) def test_node_application_promise_handler_write_after_end(self): self.load('promise_handler') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'X-Write-Call': '1', - 'Connection': 'close' - }, body='callback')['status'], 200, - 'promise handler request write after end') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'X-Write-Call': '1', + 'Connection': 'close', + }, + body='callback', + )['status'], + 200, + 'promise handler request write after end', + ) def test_node_application_promise_end(self): self.load('promise_end') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, body='end')['status'], 200, 'promise end request') - self.assertTrue(self.waitforfiles(self.testdir + '/node/callback'), - 'promise end') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + body='end', + )['status'], + 200, + 'promise end request', + ) + self.assertTrue( + self.waitforfiles(self.testdir + '/node/callback'), 'promise end' + ) def test_node_application_promise_multiple_calls(self): self.load('promise_handler') - self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, body='callback1') - - self.assertTrue(self.waitforfiles(self.testdir + '/node/callback1'), - 'promise first call') - - self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, body='callback2') - - self.assertTrue(self.waitforfiles(self.testdir + '/node/callback2'), - 'promise second call') + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + body='callback1', + ) + + self.assertTrue( + self.waitforfiles(self.testdir + '/node/callback1'), + 'promise first call', + ) + + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + body='callback2', + ) + + self.assertTrue( + self.waitforfiles(self.testdir + '/node/callback2'), + 'promise second call', + ) @unittest.expectedFailure def test_node_application_header_name_valid(self): @@ -261,28 +345,46 @@ class TestUnitNodeApplication(unit.TestUnitApplicationNode): def test_node_application_get_header_names(self): self.load('get_header_names') - self.assertListEqual(self.get()['headers']['X-Names'], - ['date', 'x-header'], 'get header names') + self.assertListEqual( + self.get()['headers']['X-Names'], + ['date', 'x-header'], + 'get header names', + ) def test_node_application_has_header(self): self.load('has_header') - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Header': 'length', - 'Connection': 'close' - })['headers']['X-Has-Header'], 'false', 'has header length') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Header': 'Date', - 'Connection': 'close' - })['headers']['X-Has-Header'], 'false', 'has header date') + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Header': 'length', + 'Connection': 'close', + } + )['headers']['X-Has-Header'], + 'false', + 'has header length', + ) + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Header': 'Date', + 'Connection': 'close', + } + )['headers']['X-Has-Header'], + 'false', + 'has header date', + ) def test_node_application_write_multiple(self): self.load('write_multiple') - self.assertEqual(self.get()['body'], 'writewrite2end', 'write multiple') + self.assertEqual( + self.get()['body'], 'writewrite2end', 'write multiple' + ) + if __name__ == '__main__': TestUnitNodeApplication.main() diff --git a/test/test_perl_application.py b/test/test_perl_application.py index b169baab..050d9a17 100644 --- a/test/test_perl_application.py +++ b/test/test_perl_application.py @@ -1,8 +1,8 @@ import unittest import unit -class TestUnitPerlApplication(unit.TestUnitApplicationPerl): +class TestUnitPerlApplication(unit.TestUnitApplicationPerl): def setUpClass(): unit.TestUnit().check_modules('perl') @@ -11,42 +11,55 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] header_server = headers.pop('Server') self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header') - self.assertEqual(headers.pop('Server-Software'), header_server, - 'server software header') + self.assertEqual( + headers.pop('Server-Software'), + header_server, + 'server software header', + ) date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') - - self.assertDictEqual(headers, { - 'Connection': 'close', - 'Content-Length': str(len(body)), - 'Content-Type': 'text/html', - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Custom-Header': 'blah', - 'Psgi-Version': '11', - 'Psgi-Url-Scheme': 'http', - 'Psgi-Multithread': '', - 'Psgi-Multiprocess': '1', - 'Psgi-Run-Once': '', - 'Psgi-Nonblocking': '', - 'Psgi-Streaming': '1' - }, 'headers') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) + + self.assertDictEqual( + headers, + { + 'Connection': 'close', + 'Content-Length': str(len(body)), + 'Content-Type': 'text/html', + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Custom-Header': 'blah', + 'Psgi-Version': '11', + 'Psgi-Url-Scheme': 'http', + 'Psgi-Multithread': '', + 'Psgi-Multiprocess': '1', + 'Psgi-Run-Once': '', + 'Psgi-Nonblocking': '', + 'Psgi-Streaming': '1', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_perl_application_query_string(self): @@ -54,8 +67,11 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): resp = self.get(url='/?var1=val1&var2=val2') - self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2', - 'Query-String header') + self.assertEqual( + resp['headers']['Query-String'], + 'var1=val1&var2=val2', + 'Query-String header', + ) def test_perl_application_query_string_empty(self): self.load('query_string') @@ -63,8 +79,9 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): resp = self.get(url='/?') self.assertEqual(resp['status'], 200, 'query string empty status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string empty') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string empty' + ) @unittest.expectedFailure def test_perl_application_query_string_absent(self): @@ -73,15 +90,17 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): resp = self.get() self.assertEqual(resp['status'], 200, 'query string absent status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string absent') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string absent' + ) @unittest.expectedFailure def test_perl_application_server_port(self): self.load('server_port') - self.assertEqual(self.get()['headers']['Server-Port'], '7080', - 'Server-Port header') + self.assertEqual( + self.get()['headers']['Server-Port'], '7080', 'Server-Port header' + ) def test_perl_application_input_read_empty(self): self.load('input_read_empty') @@ -91,15 +110,19 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): def test_perl_application_input_read_parts(self): self.load('input_read_parts') - self.assertEqual(self.post(body='0123456789')['body'], '0123456789', - 'input read parts') + self.assertEqual( + self.post(body='0123456789')['body'], + '0123456789', + 'input read parts', + ) @unittest.expectedFailure def test_perl_application_input_read_offset(self): self.load('input_read_offset') - self.assertEqual(self.post(body='0123456789')['body'], '4567', - 'read offset') + self.assertEqual( + self.post(body='0123456789')['body'], '4567', 'read offset' + ) def test_perl_application_input_copy(self): self.load('input_copy') @@ -116,13 +139,17 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): self.assertIsNotNone( self.search_in_log(r'\[error\].+Error in application'), - 'errors print') + 'errors print', + ) def test_perl_application_header_equal_names(self): self.load('header_equal_names') - self.assertListEqual(self.get()['headers']['Set-Cookie'], - ['tc=one,two,three', 'tc=four,five,six'], 'header equal names') + self.assertListEqual( + self.get()['headers']['Set-Cookie'], + ['tc=one,two,three', 'tc=four,five,six'], + 'header equal names', + ) def test_perl_application_header_pairs(self): self.load('header_pairs') @@ -160,10 +187,9 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): @unittest.expectedFailure def test_perl_application_syntax_error(self): - self.skip_alerts.extend([ - r'PSGI: Failed to parse script', - r'process \d+ exited on signal' - ]) + self.skip_alerts.extend( + [r'PSGI: Failed to parse script', r'process \d+ exited on signal'] + ) self.load('syntax_error') self.assertEqual(self.get()['status'], 500, 'syntax error') @@ -171,19 +197,27 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): def test_perl_keepalive_body(self): self.load('variables') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') @@ -194,11 +228,13 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): self.assertIsNotNone( self.search_in_log(r'\[error\].+IOFake getline\(\) \$\/ is \d+'), - 'body io fake $/ value') + 'body io fake $/ value', + ) self.assertIsNotNone( self.search_in_log(r'\[error\].+IOFake close\(\) called'), - 'body io fake close') + 'body io fake close', + ) def test_perl_delayed_response(self): self.load('delayed_response') @@ -216,5 +252,6 @@ class TestUnitPerlApplication(unit.TestUnitApplicationPerl): self.assertEqual(resp['status'], 200, 'status') self.assertEqual(resp['body'], 'Hello World!', 'body') + if __name__ == '__main__': TestUnitPerlApplication.main() diff --git a/test/test_php_application.py b/test/test_php_application.py index ddc3cad5..5d2ce118 100644 --- a/test/test_php_application.py +++ b/test/test_php_application.py @@ -2,8 +2,8 @@ import unittest import unit import re -class TestUnitPHPApplication(unit.TestUnitApplicationPHP): +class TestUnitPHPApplication(unit.TestUnitApplicationPHP): def setUpClass(): unit.TestUnit().check_modules('php') @@ -18,38 +18,51 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] header_server = headers.pop('Server') self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header') - self.assertEqual(headers.pop('Server-Software'), header_server, - 'server software header') + self.assertEqual( + headers.pop('Server-Software'), + header_server, + 'server software header', + ) date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) if 'X-Powered-By' in headers: headers.pop('X-Powered-By') headers.pop('Content-type') - self.assertDictEqual(headers, { - 'Connection': 'close', - 'Content-Length': str(len(body)), - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Custom-Header': 'blah' - }, 'headers') + self.assertDictEqual( + headers, + { + 'Connection': 'close', + 'Content-Length': str(len(body)), + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Custom-Header': 'blah', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_php_application_query_string(self): @@ -57,8 +70,11 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): resp = self.get(url='/?var1=val1&var2=val2') - self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2', - 'query string') + self.assertEqual( + resp['headers']['Query-String'], + 'var1=val1&var2=val2', + 'query string', + ) def test_php_application_query_string_empty(self): self.load('query_string') @@ -66,8 +82,9 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): resp = self.get(url='/?') self.assertEqual(resp['status'], 200, 'query string empty status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string empty') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string empty' + ) @unittest.expectedFailure def test_php_application_query_string_absent(self): @@ -76,8 +93,9 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): resp = self.get() self.assertEqual(resp['status'], 200, 'query string absent status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string absent') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string absent' + ) def test_php_application_phpinfo(self): self.load('phpinfo') @@ -93,25 +111,34 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): resp = self.get() self.assertEqual(resp['status'], 404, '404 status') - self.assertRegex(resp['body'], r'<title>404 Not Found</title>', - '404 body') + self.assertRegex( + resp['body'], r'<title>404 Not Found</title>', '404 body' + ) def test_php_application_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') @@ -133,11 +160,14 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): def test_php_application_post_variables(self): self.load('post_variables') - resp = self.post(headers={ - 'Content-Type': 'application/x-www-form-urlencoded', - 'Host': 'localhost', - 'Connection': 'close' - }, body='var1=val1&var2=') + resp = self.post( + headers={ + 'Content-Type': 'application/x-www-form-urlencoded', + 'Host': 'localhost', + 'Connection': 'close', + }, + body='var1=val1&var2=', + ) self.assertEqual(resp['headers']['X-Var-1'], 'val1', 'POST variables') self.assertEqual(resp['headers']['X-Var-2'], '1', 'POST variables 2') self.assertEqual(resp['headers']['X-Var-3'], '', 'POST variables 3') @@ -145,11 +175,13 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): def test_php_application_cookies(self): self.load('cookies') - resp = self.get(headers={ - 'Cookie': 'var=val; var2=val2', - 'Host': 'localhost', - 'Connection': 'close' - }) + resp = self.get( + headers={ + 'Cookie': 'var=val; var2=val2', + 'Host': 'localhost', + 'Connection': 'close', + } + ) self.assertEqual(resp['headers']['X-Cookie-1'], 'val', 'cookie') self.assertEqual(resp['headers']['X-Cookie-2'], 'val2', 'cookie') @@ -157,96 +189,129 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): def test_php_application_ini_precision(self): self.load('ini_precision') - self.assertNotEqual(self.get()['headers']['X-Precision'], '4', - 'ini value default') + self.assertNotEqual( + self.get()['headers']['X-Precision'], '4', 'ini value default' + ) - self.conf({"file": "ini/php.ini"}, 'applications/ini_precision/options') + self.conf( + {"file": "ini/php.ini"}, 'applications/ini_precision/options' + ) - self.assertEqual(self.get()['headers']['X-File'], - self.current_dir + '/php/ini_precision/ini/php.ini', 'ini file') - self.assertEqual(self.get()['headers']['X-Precision'], '4', 'ini value') + self.assertEqual( + self.get()['headers']['X-File'], + self.current_dir + '/php/ini_precision/ini/php.ini', + 'ini file', + ) + self.assertEqual( + self.get()['headers']['X-Precision'], '4', 'ini value' + ) @unittest.expectedFailure def test_php_application_ini_admin_user(self): self.load('ini_precision') - self.assertIn('error', self.conf({ - "user": { "precision": "4" }, - "admin": { "precision": "5" } - }, 'applications/ini_precision/options'), 'ini admin user') + self.assertIn( + 'error', + self.conf( + {"user": {"precision": "4"}, "admin": {"precision": "5"}}, + 'applications/ini_precision/options', + ), + 'ini admin user', + ) def test_php_application_ini_admin(self): self.load('ini_precision') - self.conf({ - "file": "php.ini", - "admin": { "precision": "5" } - }, 'applications/ini_precision/options') + self.conf( + {"file": "php.ini", "admin": {"precision": "5"}}, + 'applications/ini_precision/options', + ) - self.assertEqual(self.get()['headers']['X-Precision'], '5', - 'ini value admin') + self.assertEqual( + self.get()['headers']['X-Precision'], '5', 'ini value admin' + ) def test_php_application_ini_user(self): self.load('ini_precision') - self.conf({ - "file": "php.ini", - "user": { "precision": "5" } - }, 'applications/ini_precision/options') + self.conf( + {"file": "php.ini", "user": {"precision": "5"}}, + 'applications/ini_precision/options', + ) - self.assertEqual(self.get()['headers']['X-Precision'], '5', - 'ini value user') + self.assertEqual( + self.get()['headers']['X-Precision'], '5', 'ini value user' + ) def test_php_application_ini_user_2(self): self.load('ini_precision') - self.conf({"file": "ini/php.ini"}, 'applications/ini_precision/options') + self.conf( + {"file": "ini/php.ini"}, 'applications/ini_precision/options' + ) - self.assertEqual(self.get()['headers']['X-Precision'], '4', - 'ini user file') + self.assertEqual( + self.get()['headers']['X-Precision'], '4', 'ini user file' + ) - self.conf({ "precision": "5" }, - 'applications/ini_precision/options/user') + self.conf( + {"precision": "5"}, 'applications/ini_precision/options/user' + ) - self.assertEqual(self.get()['headers']['X-Precision'], '5', - 'ini value user') + self.assertEqual( + self.get()['headers']['X-Precision'], '5', 'ini value user' + ) def test_php_application_ini_set_admin(self): self.load('ini_precision') - self.conf({"admin": { "precision": "5" }}, - 'applications/ini_precision/options') + self.conf( + {"admin": {"precision": "5"}}, 'applications/ini_precision/options' + ) - self.assertEqual(self.get(url='/?precision=6')['headers']['X-Precision'], - '5', 'ini set admin') + self.assertEqual( + self.get(url='/?precision=6')['headers']['X-Precision'], + '5', + 'ini set admin', + ) def test_php_application_ini_set_user(self): self.load('ini_precision') - self.conf({"user": { "precision": "5" }}, - 'applications/ini_precision/options') + self.conf( + {"user": {"precision": "5"}}, 'applications/ini_precision/options' + ) - self.assertEqual(self.get(url='/?precision=6')['headers']['X-Precision'], - '6', 'ini set user') + self.assertEqual( + self.get(url='/?precision=6')['headers']['X-Precision'], + '6', + 'ini set user', + ) def test_php_application_ini_repeat(self): self.load('ini_precision') - self.conf({"user": { "precision": "5" }}, - 'applications/ini_precision/options') + self.conf( + {"user": {"precision": "5"}}, 'applications/ini_precision/options' + ) - self.assertEqual(self.get()['headers']['X-Precision'], '5', 'ini value') + self.assertEqual( + self.get()['headers']['X-Precision'], '5', 'ini value' + ) - self.assertEqual(self.get()['headers']['X-Precision'], '5', - 'ini value repeat') + self.assertEqual( + self.get()['headers']['X-Precision'], '5', 'ini value repeat' + ) def test_php_application_disable_functions_exec(self): self.load('time_exec') self.before_disable_functions() - self.conf({"admin": { "disable_functions": "exec" }}, - 'applications/time_exec/options') + self.conf( + {"admin": {"disable_functions": "exec"}}, + 'applications/time_exec/options', + ) body = self.get()['body'] @@ -258,80 +323,103 @@ class TestUnitPHPApplication(unit.TestUnitApplicationPHP): self.before_disable_functions() - self.conf({"admin": { "disable_functions": "exec,time" }}, - 'applications/time_exec/options') + self.conf( + {"admin": {"disable_functions": "exec,time"}}, + 'applications/time_exec/options', + ) body = self.get()['body'] self.assertNotRegex(body, r'time: \d+', 'disable_functions comma time') - self.assertNotRegex(body, r'exec: \/\w+', - 'disable_functions comma exec') + self.assertNotRegex( + body, r'exec: \/\w+', 'disable_functions comma exec' + ) def test_php_application_disable_functions_space(self): self.load('time_exec') self.before_disable_functions() - self.conf({"admin": { "disable_functions": "exec time" }}, - 'applications/time_exec/options') + self.conf( + {"admin": {"disable_functions": "exec time"}}, + 'applications/time_exec/options', + ) body = self.get()['body'] self.assertNotRegex(body, r'time: \d+', 'disable_functions space time') - self.assertNotRegex(body, r'exec: \/\w+', - 'disable_functions space exec') + self.assertNotRegex( + body, r'exec: \/\w+', 'disable_functions space exec' + ) def test_php_application_disable_functions_user(self): self.load('time_exec') self.before_disable_functions() - self.conf({"user": { "disable_functions": "exec" }}, - 'applications/time_exec/options') + self.conf( + {"user": {"disable_functions": "exec"}}, + 'applications/time_exec/options', + ) body = self.get()['body'] self.assertRegex(body, r'time: \d+', 'disable_functions user time') - self.assertNotRegex(body, r'exec: \/\w+', 'disable_functions user exec') + self.assertNotRegex( + body, r'exec: \/\w+', 'disable_functions user exec' + ) def test_php_application_disable_functions_nonexistent(self): self.load('time_exec') self.before_disable_functions() - self.conf({"admin": { "disable_functions": "blah" }}, - 'applications/time_exec/options') + self.conf( + {"admin": {"disable_functions": "blah"}}, + 'applications/time_exec/options', + ) body = self.get()['body'] - self.assertRegex(body, r'time: \d+', - 'disable_functions nonexistent time') - self.assertRegex(body, r'exec: \/\w+', - 'disable_functions nonexistent exec') + self.assertRegex( + body, r'time: \d+', 'disable_functions nonexistent time' + ) + self.assertRegex( + body, r'exec: \/\w+', 'disable_functions nonexistent exec' + ) def test_php_application_disable_classes(self): self.load('date_time') - self.assertRegex(self.get()['body'], r'012345', - 'disable_classes before') + self.assertRegex( + self.get()['body'], r'012345', 'disable_classes before' + ) - self.conf({"admin": { "disable_classes": "DateTime" }}, - 'applications/date_time/options') + self.conf( + {"admin": {"disable_classes": "DateTime"}}, + 'applications/date_time/options', + ) - self.assertNotRegex(self.get()['body'], r'012345', - 'disable_classes before') + self.assertNotRegex( + self.get()['body'], r'012345', 'disable_classes before' + ) def test_php_application_disable_classes_user(self): self.load('date_time') - self.assertRegex(self.get()['body'], r'012345', - 'disable_classes before') + self.assertRegex( + self.get()['body'], r'012345', 'disable_classes before' + ) + + self.conf( + {"user": {"disable_classes": "DateTime"}}, + 'applications/date_time/options', + ) - self.conf({"user": { "disable_classes": "DateTime" }}, - 'applications/date_time/options') + self.assertNotRegex( + self.get()['body'], r'012345', 'disable_classes before' + ) - self.assertNotRegex(self.get()['body'], r'012345', - 'disable_classes before') if __name__ == '__main__': TestUnitPHPApplication.main() diff --git a/test/test_php_basic.py b/test/test_php_basic.py index 1ea46c91..7ad2cae1 100644 --- a/test/test_php_basic.py +++ b/test/test_php_basic.py @@ -1,27 +1,23 @@ import unittest import unit -class TestUnitPHPBasic(unit.TestUnitControl): +class TestUnitPHPBasic(unit.TestUnitControl): def setUpClass(): unit.TestUnit().check_modules('php') conf_app = { "app": { "type": "php", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "root": "/app", - "index": "index.php" + "index": "index.php", } } conf_basic = { - "listeners": { - "*:7080": { - "application": "app" - } - }, - "applications": conf_app + "listeners": {"*:7080": {"application": "app"}}, + "applications": conf_app, } def test_php_get_applications(self): @@ -30,113 +26,146 @@ class TestUnitPHPBasic(unit.TestUnitControl): conf = self.conf_get() self.assertEqual(conf['listeners'], {}, 'listeners') - self.assertEqual(conf['applications'], + self.assertEqual( + conf['applications'], { "app": { "type": "php", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "root": "/app", - "index": "index.php" + "index": "index.php", } - }, - 'applications') + }, + 'applications', + ) def test_php_get_applications_prefix(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications'), + self.assertEqual( + self.conf_get('applications'), { "app": { "type": "php", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "root": "/app", - "index": "index.php" + "index": "index.php", } }, - 'applications prefix') + 'applications prefix', + ) def test_php_get_applications_prefix_2(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications/app'), + self.assertEqual( + self.conf_get('applications/app'), { "type": "php", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "root": "/app", - "index": "index.php" + "index": "index.php", }, - 'applications prefix 2') + 'applications prefix 2', + ) def test_php_get_applications_prefix_3(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications/app/type'), 'php', - 'type') - self.assertEqual(self.conf_get('applications/app/processes/spare'), 0, - 'spare processes') + self.assertEqual(self.conf_get('applications/app/type'), 'php', 'type') + self.assertEqual( + self.conf_get('applications/app/processes/spare'), + 0, + 'spare processes', + ) def test_php_get_listeners(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get()['listeners'], - {"*:7080":{"application":"app"}}, 'listeners') + self.assertEqual( + self.conf_get()['listeners'], + {"*:7080": {"application": "app"}}, + 'listeners', + ) def test_php_get_listeners_prefix(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get('listeners'), - {"*:7080":{"application":"app"}}, 'listeners prefix') + self.assertEqual( + self.conf_get('listeners'), + {"*:7080": {"application": "app"}}, + 'listeners prefix', + ) def test_php_get_listeners_prefix_2(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get('listeners/*:7080'), - {"application":"app"}, 'listeners prefix 2') + self.assertEqual( + self.conf_get('listeners/*:7080'), + {"application": "app"}, + 'listeners prefix 2', + ) def test_php_change_listener(self): self.conf(self.conf_basic) - self.conf({"*:7081":{"application":"app"}}, 'listeners') + self.conf({"*:7081": {"application": "app"}}, 'listeners') - self.assertEqual(self.conf_get('listeners'), - {"*:7081": {"application":"app"}}, 'change listener') + self.assertEqual( + self.conf_get('listeners'), + {"*:7081": {"application": "app"}}, + 'change listener', + ) def test_php_add_listener(self): self.conf(self.conf_basic) - self.conf({"application":"app"}, 'listeners/*:7082') + self.conf({"application": "app"}, 'listeners/*:7082') - self.assertEqual(self.conf_get('listeners'), + self.assertEqual( + self.conf_get('listeners'), { - "*:7080": { - "application": "app" - }, - "*:7082": { - "application": "app" - } + "*:7080": {"application": "app"}, + "*:7082": {"application": "app"}, }, - 'add listener') + 'add listener', + ) def test_php_change_application(self): self.conf(self.conf_basic) self.conf('30', 'applications/app/processes/max') - self.assertEqual(self.conf_get('applications/app/processes/max'), 30, - 'change application max') + self.assertEqual( + self.conf_get('applications/app/processes/max'), + 30, + 'change application max', + ) self.conf('"/www"', 'applications/app/root') - self.assertEqual(self.conf_get('applications/app/root'), '/www', - 'change application root') + self.assertEqual( + self.conf_get('applications/app/root'), + '/www', + 'change application root', + ) def test_php_delete(self): self.conf(self.conf_basic) - self.assertIn('error', self.conf_delete('applications/app'), - 'delete app before listener') - self.assertIn('success', self.conf_delete('listeners/*:7080'), - 'delete listener') - self.assertIn('success', self.conf_delete('applications/app'), - 'delete app after listener') - self.assertIn('error', self.conf_delete('applications/app'), - 'delete app again') + self.assertIn( + 'error', + self.conf_delete('applications/app'), + 'delete app before listener', + ) + self.assertIn( + 'success', self.conf_delete('listeners/*:7080'), 'delete listener' + ) + self.assertIn( + 'success', + self.conf_delete('applications/app'), + 'delete app after listener', + ) + self.assertIn( + 'error', self.conf_delete('applications/app'), 'delete app again' + ) + if __name__ == '__main__': TestUnitPHPBasic.main() diff --git a/test/test_python_application.py b/test/test_python_application.py index 1e8d5cc5..330bd60f 100644 --- a/test/test_python_application.py +++ b/test/test_python_application.py @@ -2,8 +2,8 @@ import time import unittest import unit -class TestUnitPythonApplication(unit.TestUnitApplicationPython): +class TestUnitPythonApplication(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') @@ -12,40 +12,53 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] header_server = headers.pop('Server') self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header') - self.assertEqual(headers.pop('Server-Software'), header_server, - 'server software header') + self.assertEqual( + headers.pop('Server-Software'), + header_server, + 'server software header', + ) date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') - - self.assertDictEqual(headers, { - 'Connection': 'close', - 'Content-Length': str(len(body)), - 'Content-Type': 'text/html', - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Custom-Header': 'blah', - 'Wsgi-Version': '(1, 0)', - 'Wsgi-Url-Scheme': 'http', - 'Wsgi-Multithread': 'False', - 'Wsgi-Multiprocess': 'True', - 'Wsgi-Run-Once': 'False' - }, 'headers') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) + + self.assertDictEqual( + headers, + { + 'Connection': 'close', + 'Content-Length': str(len(body)), + 'Content-Type': 'text/html', + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Custom-Header': 'blah', + 'Wsgi-Version': '(1, 0)', + 'Wsgi-Url-Scheme': 'http', + 'Wsgi-Multithread': 'False', + 'Wsgi-Multiprocess': 'True', + 'Wsgi-Run-Once': 'False', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_python_application_query_string(self): @@ -53,8 +66,11 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): resp = self.get(url='/?var1=val1&var2=val2') - self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2', - 'Query-String header') + self.assertEqual( + resp['headers']['Query-String'], + 'var1=val1&var2=val2', + 'Query-String header', + ) def test_python_application_query_string_empty(self): self.load('query_string') @@ -62,8 +78,9 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): resp = self.get(url='/?') self.assertEqual(resp['status'], 200, 'query string empty status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string empty') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string empty' + ) def test_python_application_query_string_absent(self): self.load('query_string') @@ -71,71 +88,87 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): resp = self.get() self.assertEqual(resp['status'], 200, 'query string absent status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string absent') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string absent' + ) @unittest.expectedFailure def test_python_application_server_port(self): self.load('server_port') - self.assertEqual(self.get()['headers']['Server-Port'], '7080', - 'Server-Port header') + self.assertEqual( + self.get()['headers']['Server-Port'], '7080', 'Server-Port header' + ) def test_python_application_204_transfer_encoding(self): self.load('204_no_content') - self.assertNotIn('Transfer-Encoding', self.get()['headers'], - '204 header transfer encoding') + self.assertNotIn( + 'Transfer-Encoding', + self.get()['headers'], + '204 header transfer encoding', + ) def test_python_application_ctx_iter_atexit(self): self.load('ctx_iter_atexit') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + body='0123456789', + ) self.assertEqual(resp['status'], 200, 'ctx iter status') self.assertEqual(resp['body'], '0123456789', 'ctx iter body') - self.conf({ - "listeners": {}, - "applications": {} - }) + self.conf({"listeners": {}, "applications": {}}) self.stop() time.sleep(0.2) - self.assertIsNotNone(self.search_in_log(r'RuntimeError'), - 'ctx iter atexit') + self.assertIsNotNone( + self.search_in_log(r'RuntimeError'), 'ctx iter atexit' + ) def test_python_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') def test_python_keepalive_reconfigure(self): - self.skip_alerts.extend([ - r'pthread_mutex.+failed', - r'failed to apply', - r'process \d+ exited on signal' - ]) + self.skip_alerts.extend( + [ + r'pthread_mutex.+failed', + r'failed to apply', + r'process \d+ exited on signal', + ] + ) self.load('mirror') body = '0123456789' @@ -143,74 +176,109 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): socks = [] for i in range(conns): - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body=body) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body=body, + ) self.assertEqual(resp['body'], body, 'keep-alive open') - self.assertIn('success', self.conf({ - "spare": i % 4, - "max": (i % 4) + 1 - }, 'applications/mirror/processes'), 'reconfigure') + self.assertIn( + 'success', + self.conf( + str(i + 1), + 'applications/mirror/processes', + ), + 'reconfigure', + ) socks.append(sock) for i in range(conns): - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, sock=socks[i], body=body) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + sock=socks[i], + body=body, + ) self.assertEqual(resp['body'], body, 'keep-alive request') - self.assertIn('success', self.conf({ - "spare": i % 4, - "max": (i % 4) + 1 - }, 'applications/mirror/processes'), 'reconfigure 2') + self.assertIn( + 'success', + self.conf( + str(i + 1), + 'applications/mirror/processes', + ), + 'reconfigure 2', + ) for i in range(conns): - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=socks[i], body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=socks[i], + body=body, + ) self.assertEqual(resp['body'], body, 'keep-alive close') - self.assertIn('success', self.conf({ - "spare": i % 4, - "max": (i % 4) + 1 - }, 'applications/mirror/processes'), 'reconfigure 3') + self.assertIn( + 'success', + self.conf( + str(i + 1), + 'applications/mirror/processes', + ), + 'reconfigure 3', + ) def test_python_keepalive_reconfigure_2(self): self.load('mirror') body = '0123456789' - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body=body) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body=body, + ) self.assertEqual(resp['body'], body, 'reconfigure 2 keep-alive 1') self.load('empty') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, start=True, sock=sock, body=body) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + start=True, + sock=sock, + body=body, + ) self.assertEqual(resp['status'], 200, 'reconfigure 2 keep-alive 2') self.assertEqual(resp['body'], '', 'reconfigure 2 keep-alive 2 body') - self.assertIn('success', self.conf({ - "listeners": {}, - "applications": {} - }), 'reconfigure 2 clear configuration') + self.assertIn( + 'success', + self.conf({"listeners": {}, "applications": {}}), + 'reconfigure 2 clear configuration', + ) resp = self.get(sock=sock) @@ -219,18 +287,27 @@ class TestUnitPythonApplication(unit.TestUnitApplicationPython): def test_python_keepalive_reconfigure_3(self): self.load('empty') - (resp, sock) = self.http(b"""GET / HTTP/1.1 -""", start=True, raw=True) - - self.assertIn('success', self.conf({ - "listeners": {}, - "applications": {} - }), 'reconfigure 3 clear configuration') - - resp = self.http(b"""Host: localhost + (resp, sock) = self.http( + b"""GET / HTTP/1.1 +""", + start=True, + raw=True, + ) + + self.assertIn( + 'success', + self.conf({"listeners": {}, "applications": {}}), + 'reconfigure 3 clear configuration', + ) + + resp = self.http( + b"""Host: localhost Connection: close -""", sock=sock, raw=True) +""", + sock=sock, + raw=True, + ) self.assertEqual(resp['status'], 200, 'reconfigure 3') @@ -239,10 +316,7 @@ Connection: close self.get() - self.conf({ - "listeners": {}, - "applications": {} - }) + self.conf({"listeners": {}, "applications": {}}) self.stop() @@ -267,35 +341,47 @@ Connection: close body = '0123456789' - resp = self.post(headers={ - 'Host': 'localhost', - 'Input-Length': '5', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Input-Length': '5', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['body'], body[:5], 'input read length lt body') - resp = self.post(headers={ - 'Host': 'localhost', - 'Input-Length': '15', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Input-Length': '15', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['body'], body, 'input read length gt body') - resp = self.post(headers={ - 'Host': 'localhost', - 'Input-Length': '0', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Input-Length': '0', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['body'], '', 'input read length zero') - resp = self.post(headers={ - 'Host': 'localhost', - 'Input-Length': '-1', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Input-Length': '-1', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['body'], body, 'input read length negative') @@ -309,7 +395,8 @@ Connection: close self.assertIsNotNone( self.search_in_log(r'\[error\].+Error in application\.'), - 'errors write') + 'errors write', + ) def test_python_application_body_array(self): self.load('body_array') @@ -349,8 +436,9 @@ Connection: close self.stop() - self.assertIsNotNone(self.search_in_log(r'Close called\.'), - 'close error') + self.assertIsNotNone( + self.search_in_log(r'Close called\.'), 'close error' + ) def test_python_application_not_iterable(self): self.load('not_iterable') @@ -359,14 +447,18 @@ Connection: close self.stop() - self.assertIsNotNone(self.search_in_log( - r'\[error\].+the application returned not an iterable object'), - 'not iterable') + self.assertIsNotNone( + self.search_in_log( + r'\[error\].+the application returned not an iterable object' + ), + 'not iterable', + ) def test_python_application_write(self): self.load('write') self.assertEqual(self.get()['body'], '0123456789', 'write') + if __name__ == '__main__': TestUnitPythonApplication.main() diff --git a/test/test_python_basic.py b/test/test_python_basic.py index b5179dea..f2a5b9f6 100644 --- a/test/test_python_basic.py +++ b/test/test_python_basic.py @@ -1,38 +1,37 @@ import unittest import unit -class TestUnitPythonBasic(unit.TestUnitControl): +class TestUnitPythonBasic(unit.TestUnitControl): def setUpClass(): unit.TestUnit().check_modules('python') conf_app = { "app": { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module": "wsgi" + "module": "wsgi", } } conf_basic = { - "listeners": { - "*:7080": { - "application": "app" - } - }, - "applications": conf_app + "listeners": {"*:7080": {"application": "app"}}, + "applications": conf_app, } def test_python_get_empty(self): - self.assertEqual(self.conf_get(), - {'listeners': {}, 'applications': {}}, 'empty') + self.assertEqual( + self.conf_get(), {'listeners': {}, 'applications': {}}, 'empty' + ) def test_python_get_prefix_listeners(self): self.assertEqual(self.conf_get('listeners'), {}, 'listeners prefix') def test_python_get_prefix_applications(self): - self.assertEqual(self.conf_get('applications'), {}, 'applications prefix') + self.assertEqual( + self.conf_get('applications'), {}, 'applications prefix' + ) def test_python_get_applications(self): self.conf(self.conf_app, 'applications') @@ -40,113 +39,146 @@ class TestUnitPythonBasic(unit.TestUnitControl): conf = self.conf_get() self.assertEqual(conf['listeners'], {}, 'listeners') - self.assertEqual(conf['applications'], + self.assertEqual( + conf['applications'], { "app": { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module": "wsgi" + "module": "wsgi", } }, - 'applications') + 'applications', + ) def test_python_get_applications_prefix(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications'), + self.assertEqual( + self.conf_get('applications'), { "app": { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module":"wsgi" + "module": "wsgi", } }, - 'applications prefix') + 'applications prefix', + ) def test_python_get_applications_prefix_2(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications/app'), + self.assertEqual( + self.conf_get('applications/app'), { "type": "python", - "processes": { "spare": 0 }, + "processes": {"spare": 0}, "path": "/app", - "module": "wsgi" + "module": "wsgi", }, - 'applications prefix 2') + 'applications prefix 2', + ) def test_python_get_applications_prefix_3(self): self.conf(self.conf_app, 'applications') - self.assertEqual(self.conf_get('applications/app/type'), 'python', - 'type') - self.assertEqual(self.conf_get('applications/app/processes/spare'), 0, - 'spare') + self.assertEqual( + self.conf_get('applications/app/type'), 'python', 'type' + ) + self.assertEqual( + self.conf_get('applications/app/processes/spare'), 0, 'spare' + ) def test_python_get_listeners(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get()['listeners'], - {"*:7080":{"application":"app"}}, 'listeners') + self.assertEqual( + self.conf_get()['listeners'], + {"*:7080": {"application": "app"}}, + 'listeners', + ) def test_python_get_listeners_prefix(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get('listeners'), - {"*:7080":{"application":"app"}}, 'listeners prefix') + self.assertEqual( + self.conf_get('listeners'), + {"*:7080": {"application": "app"}}, + 'listeners prefix', + ) def test_python_get_listeners_prefix_2(self): self.conf(self.conf_basic) - self.assertEqual(self.conf_get('listeners/*:7080'), - {"application":"app"}, 'listeners prefix 2') + self.assertEqual( + self.conf_get('listeners/*:7080'), + {"application": "app"}, + 'listeners prefix 2', + ) def test_python_change_listener(self): self.conf(self.conf_basic) - self.conf({"*:7081":{"application":"app"}}, 'listeners') + self.conf({"*:7081": {"application": "app"}}, 'listeners') - self.assertEqual(self.conf_get('listeners'), - {"*:7081": {"application":"app"}}, 'change listener') + self.assertEqual( + self.conf_get('listeners'), + {"*:7081": {"application": "app"}}, + 'change listener', + ) def test_python_add_listener(self): self.conf(self.conf_basic) - self.conf({"application":"app"}, 'listeners/*:7082') + self.conf({"application": "app"}, 'listeners/*:7082') - self.assertEqual(self.conf_get('listeners'), + self.assertEqual( + self.conf_get('listeners'), { - "*:7080": { - "application": "app" - }, - "*:7082": { - "application": "app" - } + "*:7080": {"application": "app"}, + "*:7082": {"application": "app"}, }, - 'add listener') + 'add listener', + ) def test_python_change_application(self): self.conf(self.conf_basic) self.conf('30', 'applications/app/processes/max') - self.assertEqual(self.conf_get('applications/app/processes/max'), 30, - 'change application max') + self.assertEqual( + self.conf_get('applications/app/processes/max'), + 30, + 'change application max', + ) self.conf('"/www"', 'applications/app/path') - self.assertEqual(self.conf_get('applications/app/path'), '/www', - 'change application path') + self.assertEqual( + self.conf_get('applications/app/path'), + '/www', + 'change application path', + ) def test_python_delete(self): self.conf(self.conf_basic) - self.assertIn('error', self.conf_delete('applications/app'), - 'delete app before listener') - self.assertIn('success', self.conf_delete('listeners/*:7080'), - 'delete listener') - self.assertIn('success', self.conf_delete('applications/app'), - 'delete app after listener') - self.assertIn('error', self.conf_delete('applications/app'), - 'delete app again') + self.assertIn( + 'error', + self.conf_delete('applications/app'), + 'delete app before listener', + ) + self.assertIn( + 'success', self.conf_delete('listeners/*:7080'), 'delete listener' + ) + self.assertIn( + 'success', + self.conf_delete('applications/app'), + 'delete app after listener', + ) + self.assertIn( + 'error', self.conf_delete('applications/app'), 'delete app again' + ) + if __name__ == '__main__': TestUnitPythonBasic.main() diff --git a/test/test_python_environment.py b/test/test_python_environment.py index 71e4d5b7..8ab82089 100644 --- a/test/test_python_environment.py +++ b/test/test_python_environment.py @@ -1,128 +1,181 @@ import unittest import unit -class TestUnitPythonEnvironment(unit.TestUnitApplicationPython): +class TestUnitPythonEnvironment(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') def test_python_environment_name_null(self): self.load('environment') - self.assertIn('error', self.conf({ - "va\0r": "val1" - }, 'applications/environment/environment'), 'name null') + self.assertIn( + 'error', + self.conf( + {"va\0r": "val1"}, 'applications/environment/environment' + ), + 'name null', + ) def test_python_environment_name_equals(self): self.load('environment') - self.assertIn('error', self.conf({ - "var=": "val1" - }, 'applications/environment/environment'), 'name equals') + self.assertIn( + 'error', + self.conf( + {"var=": "val1"}, 'applications/environment/environment' + ), + 'name equals', + ) def test_python_environment_value_null(self): self.load('environment') - self.assertIn('error', self.conf({ - "var": "\0val" - }, 'applications/environment/environment'), 'value null') + self.assertIn( + 'error', + self.conf( + {"var": "\0val"}, 'applications/environment/environment' + ), + 'value null', + ) def test_python_environment_update(self): self.load('environment') - self.conf({ - "var": "val1" - }, 'applications/environment/environment') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var', - 'Connection': 'close' - })['body'], 'val1,', 'set') - - self.conf({ - "var": "val2" - }, 'applications/environment/environment') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var', - 'Connection': 'close' - })['body'], 'val2,', 'update') + self.conf({"var": "val1"}, 'applications/environment/environment') + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var', + 'Connection': 'close', + } + )['body'], + 'val1,', + 'set', + ) + + self.conf({"var": "val2"}, 'applications/environment/environment') + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var', + 'Connection': 'close', + } + )['body'], + 'val2,', + 'update', + ) def test_python_environment_replace(self): self.load('environment') - self.conf({ - "var1": "val1" - }, 'applications/environment/environment') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var1', - 'Connection': 'close' - })['body'], 'val1,', 'set') - - self.conf({ - "var2": "val2" - }, 'applications/environment/environment') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var1,var2', - 'Connection': 'close' - })['body'], 'val2,', 'replace') + self.conf({"var1": "val1"}, 'applications/environment/environment') + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var1', + 'Connection': 'close', + } + )['body'], + 'val1,', + 'set', + ) + + self.conf({"var2": "val2"}, 'applications/environment/environment') + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var1,var2', + 'Connection': 'close', + } + )['body'], + 'val2,', + 'replace', + ) def test_python_environment_clear(self): self.load('environment') - self.conf({ - "var1": "val1", - "var2": "val2" - }, 'applications/environment/environment') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var1,var2', - 'Connection': 'close' - })['body'], 'val1,val2,', 'set') + self.conf( + {"var1": "val1", "var2": "val2"}, + 'applications/environment/environment', + ) + + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var1,var2', + 'Connection': 'close', + } + )['body'], + 'val1,val2,', + 'set', + ) self.conf({}, 'applications/environment/environment') - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'var1,var2', - 'Connection': 'close' - })['body'], '', 'clear') + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'var1,var2', + 'Connection': 'close', + } + )['body'], + '', + 'clear', + ) def test_python_environment_replace_default(self): self.load('environment') - pwd_default = self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'PWD', - 'Connection': 'close' - })['body'] + pwd_default = self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'PWD', + 'Connection': 'close', + } + )['body'] self.assertGreater(len(pwd_default), 1, 'get default') - self.conf({ - "PWD": "new/pwd" - }, 'applications/environment/environment') + self.conf({"PWD": "new/pwd"}, 'applications/environment/environment') - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'PWD', - 'Connection': 'close' - })['body'], 'new/pwd,', 'replace default') + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'PWD', + 'Connection': 'close', + } + )['body'], + 'new/pwd,', + 'replace default', + ) self.conf({}, 'applications/environment/environment') - self.assertEqual(self.get(headers={ - 'Host': 'localhost', - 'X-Variables': 'PWD', - 'Connection': 'close' - })['body'], pwd_default, 'restore default') + self.assertEqual( + self.get( + headers={ + 'Host': 'localhost', + 'X-Variables': 'PWD', + 'Connection': 'close', + } + )['body'], + pwd_default, + 'restore default', + ) + if __name__ == '__main__': TestUnitPythonEnvironment.main() diff --git a/test/test_python_procman.py b/test/test_python_procman.py index 2efe59c0..f4f53678 100644 --- a/test/test_python_procman.py +++ b/test/test_python_procman.py @@ -4,8 +4,8 @@ import subprocess import unittest import unit -class TestUnitPythonProcman(unit.TestUnitApplicationPython): +class TestUnitPythonProcman(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') @@ -29,55 +29,88 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): def test_python_processes_access(self): self.conf('1', 'applications/' + self.app_name + '/processes') - self.assertIn('error', self.conf_get('/applications/' + self.app_name + - '/processes/max'), 'max no access') - self.assertIn('error', self.conf_get('/applications/' + self.app_name + - '/processes/spare'), 'spare no access') - self.assertIn('error', self.conf_get('/applications/' + self.app_name + - '/processes/idle_timeout'), 'idle_timeout no access') + self.assertIn( + 'error', + self.conf_get('/applications/' + self.app_name + '/processes/max'), + 'max no access', + ) + self.assertIn( + 'error', + self.conf_get( + '/applications/' + self.app_name + '/processes/spare' + ), + 'spare no access', + ) + self.assertIn( + 'error', + self.conf_get( + '/applications/' + self.app_name + '/processes/idle_timeout' + ), + 'idle_timeout no access', + ) def test_python_processes_spare_negative(self): - self.assertIn('error', self.conf({ - "spare": -1 - }, 'applications/' + self.app_name + '/processes'), 'negative spare') + self.assertIn( + 'error', + self.conf( + {"spare": -1}, 'applications/' + self.app_name + '/processes' + ), + 'negative spare', + ) def test_python_processes_max_negative(self): - self.assertIn('error', self.conf({ - "max": -1 - }, 'applications/' + self.app_name + '/processes'), 'negative max') + self.assertIn( + 'error', + self.conf( + {"max": -1}, 'applications/' + self.app_name + '/processes' + ), + 'negative max', + ) def test_python_processes_idle_timeout_negative(self): - self.assertIn('error', self.conf({ - "idle_timeout": -1 - }, 'applications/' + self.app_name + '/processes'), - 'negative idle_timeout') + self.assertIn( + 'error', + self.conf( + {"idle_timeout": -1}, + 'applications/' + self.app_name + '/processes', + ), + 'negative idle_timeout', + ) def test_python_processes_spare_gt_max_default(self): - self.assertIn('error', self.conf({"spare": 2}, - 'applications/' + self.app_name + '/processes'), - 'spare greater than max default') + self.assertIn( + 'error', + self.conf( + {"spare": 2}, 'applications/' + self.app_name + '/processes' + ), + 'spare greater than max default', + ) def test_python_processes_spare_gt_max(self): - self.assertIn('error', self.conf({ - "spare": 2, - "max": 1, - "idle_timeout": 1 - }, '/applications/' + self.app_name + '/processes'), - 'spare greater than max') + self.assertIn( + 'error', + self.conf( + {"spare": 2, "max": 1, "idle_timeout": 1}, + '/applications/' + self.app_name + '/processes', + ), + 'spare greater than max', + ) def test_python_processes_max_zero(self): - self.assertIn('error', self.conf({ - "spare": 0, - "max": 0, - "idle_timeout": 1 - }, 'applications/' + self.app_name + '/processes'), 'max 0') + self.assertIn( + 'error', + self.conf( + {"spare": 0, "max": 0, "idle_timeout": 1}, + 'applications/' + self.app_name + '/processes', + ), + 'max 0', + ) def test_python_processes_idle_timeout_zero(self): - self.conf({ - "spare": 0, - "max": 2, - "idle_timeout": 0 - }, 'applications/' + self.app_name + '/processes') + self.conf( + {"spare": 0, "max": 2, "idle_timeout": 0}, + 'applications/' + self.app_name + '/processes', + ) self.get() self.assertEqual(len(self.pids_for_process()), 0, 'idle timeout 0') @@ -114,11 +147,10 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): self.assertTrue(pids.issubset(pids_new), 'prefork same processes') def test_python_ondemand(self): - self.conf({ - "spare": 0, - "max": 8, - "idle_timeout": 1 - }, 'applications/' + self.app_name + '/processes') + self.conf( + {"spare": 0, "max": 8, "idle_timeout": 1}, + 'applications/' + self.app_name + '/processes', + ) self.assertEqual(len(self.pids_for_process()), 0, 'on-demand 0') @@ -131,16 +163,17 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): time.sleep(1) - self.assertEqual(len(self.pids_for_process()), 0, 'on-demand stop idle') + self.assertEqual( + len(self.pids_for_process()), 0, 'on-demand stop idle' + ) self.stop_all() def test_python_scale_updown(self): - self.conf({ - "spare": 2, - "max": 8, - "idle_timeout": 1 - }, 'applications/' + self.app_name + '/processes') + self.conf( + {"spare": 2, "max": 8, "idle_timeout": 1}, + 'applications/' + self.app_name + '/processes', + ) pids = self.pids_for_process() self.assertEqual(len(pids), 2, 'updown 2') @@ -151,7 +184,9 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): self.assertTrue(pids.issubset(pids_new), 'updown 3 only 1 new') self.get() - self.assertSetEqual(self.pids_for_process(), pids_new, 'updown still 3') + self.assertSetEqual( + self.pids_for_process(), pids_new, 'updown still 3' + ) time.sleep(1) @@ -166,11 +201,10 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): self.stop_all() def test_python_reconfigure(self): - self.conf({ - "spare": 2, - "max": 6, - "idle_timeout": 1 - }, 'applications/' + self.app_name + '/processes') + self.conf( + {"spare": 2, "max": 6, "idle_timeout": 1}, + 'applications/' + self.app_name + '/processes', + ) pids = self.pids_for_process() self.assertEqual(len(pids), 2, 'reconf 2') @@ -191,11 +225,10 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): self.stop_all() def test_python_idle_timeout(self): - self.conf({ - "spare": 0, - "max": 6, - "idle_timeout": 2 - }, 'applications/' + self.app_name + '/processes') + self.conf( + {"spare": 0, "max": 6, "idle_timeout": 2}, + 'applications/' + self.app_name + '/processes', + ) self.get() pids = self.pids_for_process() @@ -209,40 +242,42 @@ class TestUnitPythonProcman(unit.TestUnitApplicationPython): pids_new = self.pids_for_process() self.assertEqual(len(pids_new), 1, 'idle timeout still 1') - self.assertSetEqual(self.pids_for_process(), pids, - 'idle timeout still 1 same pid') + self.assertSetEqual( + self.pids_for_process(), pids, 'idle timeout still 1 same pid' + ) time.sleep(1) self.assertEqual(len(self.pids_for_process()), 0, 'idle timed out') def test_python_processes_connection_keepalive(self): - self.conf({ - "spare": 0, - "max": 6, - "idle_timeout": 2 - }, 'applications/' + self.app_name + '/processes') - - (resp, sock) = self.get(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive' - }, start=True, read_timeout=1) - self.assertEqual(len(self.pids_for_process()), 1, - 'keepalive connection 1') + self.conf( + {"spare": 0, "max": 6, "idle_timeout": 2}, + 'applications/' + self.app_name + '/processes', + ) + + (resp, sock) = self.get( + headers={'Host': 'localhost', 'Connection': 'keep-alive'}, + start=True, + read_timeout=1, + ) + self.assertEqual( + len(self.pids_for_process()), 1, 'keepalive connection 1' + ) time.sleep(2) - self.assertEqual(len(self.pids_for_process()), 0, 'keepalive connection 0') + self.assertEqual( + len(self.pids_for_process()), 0, 'keepalive connection 0' + ) sock.close() def stop_all(self): - self.conf({ - "listeners": {}, - "applications": {} - }) + self.conf({"listeners": {}, "applications": {}}) self.assertEqual(len(self.pids_for_process()), 0, 'stop all') + if __name__ == '__main__': TestUnitPythonProcman.main() diff --git a/test/test_routing.py b/test/test_routing.py index 07097fc8..b5c42576 100644 --- a/test/test_routing.py +++ b/test/test_routing.py @@ -1,458 +1,764 @@ import unittest import unit -class TestUnitRouting(unit.TestUnitApplicationProto): +class TestUnitRouting(unit.TestUnitApplicationProto): def setUpClass(): unit.TestUnit().check_modules('python') def setUp(self): super().setUp() - self.conf({ - "listeners": { - "*:7080": { - "pass": "routes" - } - }, - "routes": [{ - "match": { "method": "GET" }, - "action": { "pass": "applications/empty" } - }], - "applications": { - "empty": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir + '/python/empty', - "module": "wsgi" + self.conf( + { + "listeners": {"*:7080": {"pass": "routes"}}, + "routes": [ + { + "match": {"method": "GET"}, + "action": {"pass": "applications/empty"}, + } + ], + "applications": { + "empty": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/empty', + "working_directory": self.current_dir + + '/python/empty', + "module": "wsgi", + }, + "mirror": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/mirror', + "working_directory": self.current_dir + + '/python/mirror', + "module": "wsgi", + }, }, - "mirror": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/mirror', - "working_directory": self.current_dir + '/python/mirror', - "module": "wsgi" - } } - }) + ) def test_routes_match_method_positive(self): self.assertEqual(self.get()['status'], 200, 'method positive GET') self.assertEqual(self.post()['status'], 404, 'method positive POST') def test_routes_match_method_positive_many(self): - self.assertIn('success', self.conf([{ - "match": { "method": ["GET", "POST"] }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method positive many configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": ["GET", "POST"]}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method positive many configure', + ) self.assertEqual(self.get()['status'], 200, 'method positive many GET') - self.assertEqual(self.post()['status'], 200, - 'method positive many POST') - self.assertEqual(self.delete()['status'], 404, - 'method positive many DELETE') + self.assertEqual( + self.post()['status'], 200, 'method positive many POST' + ) + self.assertEqual( + self.delete()['status'], 404, 'method positive many DELETE' + ) def test_routes_match_method_negative(self): - self.assertIn('success', self.conf([{ - "match": { "method": "!GET" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method negative configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "!GET"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method negative configure', + ) self.assertEqual(self.get()['status'], 404, 'method negative GET') self.assertEqual(self.post()['status'], 200, 'method negative POST') def test_routes_match_method_negative_many(self): - self.assertIn('success', self.conf([{ - "match": { "method": ["!GET", "!POST"] }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method negative many configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": ["!GET", "!POST"]}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method negative many configure', + ) self.assertEqual(self.get()['status'], 404, 'method negative many GET') - self.assertEqual(self.post()['status'], 404, - 'method negative many POST') - self.assertEqual(self.delete()['status'], 200, - 'method negative many DELETE') + self.assertEqual( + self.post()['status'], 404, 'method negative many POST' + ) + self.assertEqual( + self.delete()['status'], 200, 'method negative many DELETE' + ) def test_routes_match_method_wildcard_left(self): - self.assertIn('success', self.conf([{ - "match": { "method": "*ET" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method wildcard left configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "*ET"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method wildcard left configure', + ) self.assertEqual(self.get()['status'], 200, 'method wildcard left GET') - self.assertEqual(self.post()['status'], 404, - 'method wildcard left POST') + self.assertEqual( + self.post()['status'], 404, 'method wildcard left POST' + ) def test_routes_match_method_wildcard_right(self): - self.assertIn('success', self.conf([{ - "match": { "method": "GE*" }, - "action": { "pass": "applications/empty"} - }], 'routes'), 'method wildcard right configure') - - self.assertEqual(self.get()['status'], 200, - 'method wildcard right GET') - self.assertEqual(self.post()['status'], 404, - 'method wildcard right POST') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "GE*"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method wildcard right configure', + ) + + self.assertEqual( + self.get()['status'], 200, 'method wildcard right GET' + ) + self.assertEqual( + self.post()['status'], 404, 'method wildcard right POST' + ) def test_routes_match_method_wildcard_left_right(self): - self.assertIn('success', self.conf([{ - "match": { "method": "*GET*" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method wildcard left right configure') - - self.assertEqual(self.get()['status'], 200, - 'method wildcard right GET') - self.assertEqual(self.post()['status'], 404, - 'method wildcard right POST') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "*GET*"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method wildcard left right configure', + ) + + self.assertEqual( + self.get()['status'], 200, 'method wildcard right GET' + ) + self.assertEqual( + self.post()['status'], 404, 'method wildcard right POST' + ) def test_routes_match_method_wildcard(self): - self.assertIn('success', self.conf([{ - "match": { "method": "*" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method wildcard configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "*"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method wildcard configure', + ) self.assertEqual(self.get()['status'], 200, 'method wildcard') def test_routes_match_method_case_insensitive(self): - self.assertIn('success', self.conf([{ - "match": { "method": "get" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'method case insensitive configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "get"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'method case insensitive configure', + ) self.assertEqual(self.get()['status'], 200, 'method case insensitive') def test_routes_absent(self): - self.conf({ - "listeners": { - "*:7081": { - "pass": "applications/empty" - } - }, - "applications": { - "empty": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir + '/python/empty', - "module": "wsgi" - } + self.conf( + { + "listeners": {"*:7081": {"pass": "applications/empty"}}, + "applications": { + "empty": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/empty', + "working_directory": self.current_dir + + '/python/empty', + "module": "wsgi", + } + }, } - }) + ) self.assertEqual(self.get(port=7081)['status'], 200, 'routes absent') def test_routes_pass_invalid(self): - self.assertIn('error', self.conf({ "pass": "routes/blah" }, - 'listeners/*:7080'), 'routes invalid') + self.assertIn( + 'error', + self.conf({"pass": "routes/blah"}, 'listeners/*:7080'), + 'routes invalid', + ) def test_route_empty(self): - self.assertIn('success', self.conf({ - "listeners": { - "*:7080": { - "pass": "routes/main" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"*:7080": {"pass": "routes/main"}}, + "routes": {"main": []}, + "applications": { + "empty": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/empty', + "working_directory": self.current_dir + + '/python/empty', + "module": "wsgi", + }, + "mirror": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/mirror', + "working_directory": self.current_dir + + '/python/mirror', + "module": "wsgi", + }, + }, } - }, - "routes": {"main": []}, - "applications": { - "empty": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir + '/python/empty', - "module": "wsgi" - }, - "mirror": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/mirror', - "working_directory": self.current_dir + '/python/mirror', - "module": "wsgi" - } - } - }), 'route empty configure') + ), + 'route empty configure', + ) self.assertEqual(self.get()['status'], 404, 'route empty') def test_routes_route_empty(self): - self.assertIn('success', self.conf({}, 'listeners'), - 'routes empty listeners configure') + self.assertIn( + 'success', + self.conf({}, 'listeners'), + 'routes empty listeners configure', + ) - self.assertIn('success', self.conf({}, 'routes'), - 'routes empty configure') + self.assertIn( + 'success', self.conf({}, 'routes'), 'routes empty configure' + ) def test_routes_route_match_absent(self): - self.assertIn('success', self.conf([{ - "action": { "pass": "applications/empty" } - }], 'routes'), 'route match absent configure') + self.assertIn( + 'success', + self.conf([{"action": {"pass": "applications/empty"}}], 'routes'), + 'route match absent configure', + ) self.assertEqual(self.get()['status'], 200, 'route match absent') def test_routes_route_action_absent(self): self.skip_alerts.append(r'failed to apply new conf') - self.assertIn('error', self.conf([{ - "match": { "method": "GET" } - }], 'routes'), 'route pass absent configure') + self.assertIn( + 'error', + self.conf([{"match": {"method": "GET"}}], 'routes'), + 'route pass absent configure', + ) def test_routes_route_pass_absent(self): self.skip_alerts.append(r'failed to apply new conf') - self.assertIn('error', self.conf([{ - "match": { "method": "GET" }, - "action": {} - }], 'routes'), 'route pass absent configure') + self.assertIn( + 'error', + self.conf([{"match": {"method": "GET"}, "action": {}}], 'routes'), + 'route pass absent configure', + ) def test_routes_rules_two(self): - self.assertIn('success', self.conf([{ - "match": { "method": "GET" }, - "action": { "pass": "applications/empty" } - }, - { - "match": { "method": "POST" }, - "action": { "pass": "applications/mirror" } - }], 'routes'), 'rules two configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"method": "GET"}, + "action": {"pass": "applications/empty"}, + }, + { + "match": {"method": "POST"}, + "action": {"pass": "applications/mirror"}, + }, + ], + 'routes', + ), + 'rules two configure', + ) self.assertEqual(self.get()['status'], 200, 'rules two match first') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Connection': 'close' - }, body='X')['status'], 200, 'rules two match second') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Connection': 'close', + }, + body='X', + )['status'], + 200, + 'rules two match second', + ) def test_routes_two(self): - self.assertIn('success', self.conf({ - "listeners": { - "*:7080": { - "pass": "routes/first" - } - }, - "routes": { - "first": [{ - "match": { "method": "GET" }, - "action": { "pass": "routes/second" } - }], - "second": [{ - "match": { "host": "localhost" }, - "action": { "pass": "applications/empty" } - }], - }, - "applications": { - "empty": { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/empty', - "working_directory": self.current_dir + '/python/empty', - "module": "wsgi" + self.assertIn( + 'success', + self.conf( + { + "listeners": {"*:7080": {"pass": "routes/first"}}, + "routes": { + "first": [ + { + "match": {"method": "GET"}, + "action": {"pass": "routes/second"}, + } + ], + "second": [ + { + "match": {"host": "localhost"}, + "action": {"pass": "applications/empty"}, + } + ], + }, + "applications": { + "empty": { + "type": "python", + "processes": {"spare": 0}, + "path": self.current_dir + '/python/empty', + "working_directory": self.current_dir + + '/python/empty', + "module": "wsgi", + } + }, } - } - }), 'routes two configure') + ), + 'routes two configure', + ) self.assertEqual(self.get()['status'], 200, 'routes two') def test_routes_match_host_positive(self): - self.assertIn('success', self.conf([{ - "match": { "host": "localhost" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host positive configure') - - self.assertEqual(self.get()['status'], 200, - 'match host positive localhost') - - self.assertEqual(self.get(headers={'Connection': 'close'})['status'], - 404, 'match host positive empty') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost.', - 'Connection': 'close' - })['status'], 200, 'match host positive trailing dot') - - self.assertEqual(self.get(headers={ - 'Host': 'www.localhost', - 'Connection': 'close' - })['status'], 404, 'match host positive www.localhost') - - self.assertEqual(self.get(headers={ - 'Host': 'localhost1', - 'Connection': 'close' - })['status'], 404, 'match host positive localhost1') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com', - 'Connection': 'close' - })['status'], 404, 'match host positive example.com') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": "localhost"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host positive configure', + ) + + self.assertEqual( + self.get()['status'], 200, 'match host positive localhost' + ) + + self.assertEqual( + self.get(headers={'Connection': 'close'})['status'], + 404, + 'match host positive empty', + ) + + self.assertEqual( + self.get(headers={'Host': 'localhost.', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host positive trailing dot', + ) + + self.assertEqual( + self.get(headers={'Host': 'www.localhost', 'Connection': 'close'})[ + 'status' + ], + 404, + 'match host positive www.localhost', + ) + + self.assertEqual( + self.get(headers={'Host': 'localhost1', 'Connection': 'close'})[ + 'status' + ], + 404, + 'match host positive localhost1', + ) + + self.assertEqual( + self.get(headers={'Host': 'example.com', 'Connection': 'close'})[ + 'status' + ], + 404, + 'match host positive example.com', + ) def test_routes_match_host_ipv4(self): - self.assertIn('success', self.conf([{ - "match": { "host": "127.0.0.1" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host ipv4 configure') - - self.assertEqual(self.get(headers={ - 'Host': '127.0.0.1', - 'Connection': 'close' - })['status'], 200, 'match host ipv4') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": "127.0.0.1"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host ipv4 configure', + ) + + self.assertEqual( + self.get(headers={'Host': '127.0.0.1', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host ipv4', + ) def test_routes_match_host_ipv6(self): - self.assertIn('success', self.conf([{ - "match": { "host": "[::1]" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host ipv6 configure') - - self.assertEqual(self.get(headers={ - 'Host': '[::1]', - 'Connection': 'close' - })['status'], 200, 'match host ipv6') - - self.assertEqual(self.get(headers={ - 'Host': '[::1]:7080', - 'Connection': 'close' - })['status'], 200, 'match host ipv6 port') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": "[::1]"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host ipv6 configure', + ) + + self.assertEqual( + self.get(headers={'Host': '[::1]', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host ipv6', + ) + + self.assertEqual( + self.get(headers={'Host': '[::1]:7080', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host ipv6 port', + ) def test_routes_match_host_positive_many(self): - self.assertIn('success', self.conf([{ - "match": { "host": ["localhost", "example.com"] }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host positive many configure') - - self.assertEqual(self.get()['status'], 200, - 'match host positive many localhost') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com', - 'Connection': 'close' - })['status'], 200, 'match host positive many example.com') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": ["localhost", "example.com"]}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host positive many configure', + ) + + self.assertEqual( + self.get()['status'], 200, 'match host positive many localhost' + ) + + self.assertEqual( + self.get(headers={'Host': 'example.com', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host positive many example.com', + ) def test_routes_match_host_positive_and_negative(self): - self.assertIn('success', self.conf([{ - "match": { "host": ["*example.com", "!www.example.com"] }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host positive and negative configure') - - self.assertEqual(self.get()['status'], 404, - 'match host positive and negative localhost') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com', - 'Connection': 'close' - })['status'], 200, 'match host positive and negative example.com') - - self.assertEqual(self.get(headers={ - 'Host': 'www.example.com', - 'Connection': 'close' - })['status'], 404, 'match host positive and negative www.example.com') - - self.assertEqual(self.get(headers={ - 'Host': '!www.example.com', - 'Connection': 'close' - })['status'], 200, 'match host positive and negative !www.example.com') + self.assertIn( + 'success', + self.conf( + [ + { + "match": { + "host": ["*example.com", "!www.example.com"] + }, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host positive and negative configure', + ) + + self.assertEqual( + self.get()['status'], + 404, + 'match host positive and negative localhost', + ) + + self.assertEqual( + self.get(headers={'Host': 'example.com', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host positive and negative example.com', + ) + + self.assertEqual( + self.get( + headers={'Host': 'www.example.com', 'Connection': 'close'} + )['status'], + 404, + 'match host positive and negative www.example.com', + ) + + self.assertEqual( + self.get( + headers={'Host': '!www.example.com', 'Connection': 'close'} + )['status'], + 200, + 'match host positive and negative !www.example.com', + ) def test_routes_match_host_positive_and_negative_wildcard(self): - self.assertIn('success', self.conf([{ - "match": { "host": ["*example*", "!www.example*"] }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host positive and negative wildcard configure') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com', - 'Connection': 'close' - })['status'], 200, - 'match host positive and negative wildcard example.com') - - self.assertEqual(self.get(headers={ - 'Host': 'www.example.com', - 'Connection': 'close' - })['status'], 404, - 'match host positive and negative wildcard www.example.com') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": ["*example*", "!www.example*"]}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host positive and negative wildcard configure', + ) + + self.assertEqual( + self.get(headers={'Host': 'example.com', 'Connection': 'close'})[ + 'status' + ], + 200, + 'match host positive and negative wildcard example.com', + ) + + self.assertEqual( + self.get( + headers={'Host': 'www.example.com', 'Connection': 'close'} + )['status'], + 404, + 'match host positive and negative wildcard www.example.com', + ) def test_routes_match_host_case_insensitive(self): - self.assertIn('success', self.conf([{ - "match": { "host": "Example.com" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'host case insensitive configure') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com', - 'Connection': 'close' - })['status'], 200, 'host case insensitive example.com') - - self.assertEqual(self.get(headers={ - 'Host': 'EXAMPLE.COM', - 'Connection': 'close' - })['status'], 200, 'host case insensitive EXAMPLE.COM') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": "Example.com"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'host case insensitive configure', + ) + + self.assertEqual( + self.get(headers={'Host': 'example.com', 'Connection': 'close'})[ + 'status' + ], + 200, + 'host case insensitive example.com', + ) + + self.assertEqual( + self.get(headers={'Host': 'EXAMPLE.COM', 'Connection': 'close'})[ + 'status' + ], + 200, + 'host case insensitive EXAMPLE.COM', + ) def test_routes_match_host_port(self): - self.assertIn('success', self.conf([{ - "match": { "host": "example.com" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match host port configure') - - self.assertEqual(self.get(headers={ - 'Host': 'example.com:7080', - 'Connection': 'close' - })['status'], 200, 'match host port') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"host": "example.com"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match host port configure', + ) + + self.assertEqual( + self.get( + headers={'Host': 'example.com:7080', 'Connection': 'close'} + )['status'], + 200, + 'match host port', + ) def test_routes_match_uri_positive(self): - self.assertIn('success', self.conf([{ - "match": { "uri": "/" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match uri positive configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"uri": "/"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match uri positive configure', + ) self.assertEqual(self.get()['status'], 200, 'match uri positive') - self.assertEqual(self.get(url='/blah')['status'], 404, - 'match uri positive blah') - self.assertEqual(self.get(url='/#blah')['status'], 200, - 'match uri positive #blah') - self.assertEqual(self.get(url='/?var')['status'], 200, - 'match uri params') - self.assertEqual(self.get(url='//')['status'], 200, - 'match uri adjacent slashes') - self.assertEqual(self.get(url='/blah/../')['status'], 200, - 'match uri relative path') - self.assertEqual(self.get(url='/./')['status'], 200, - 'match uri relative path') + self.assertEqual( + self.get(url='/blah')['status'], 404, 'match uri positive blah' + ) + self.assertEqual( + self.get(url='/#blah')['status'], 200, 'match uri positive #blah' + ) + self.assertEqual( + self.get(url='/?var')['status'], 200, 'match uri params' + ) + self.assertEqual( + self.get(url='//')['status'], 200, 'match uri adjacent slashes' + ) + self.assertEqual( + self.get(url='/blah/../')['status'], 200, 'match uri relative path' + ) + self.assertEqual( + self.get(url='/./')['status'], 200, 'match uri relative path' + ) def test_routes_match_uri_case_sensitive(self): - self.assertIn('success', self.conf([{ - "match": { "uri": "/BLAH" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match uri case sensitive configure') - - self.assertEqual(self.get(url='/blah')['status'], 404, - 'match uri case sensitive blah') - self.assertEqual(self.get(url='/BlaH')['status'], 404, - 'match uri case sensitive BlaH') - self.assertEqual(self.get(url='/BLAH')['status'], 200, - 'match uri case sensitive BLAH') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"uri": "/BLAH"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match uri case sensitive configure', + ) + + self.assertEqual( + self.get(url='/blah')['status'], + 404, + 'match uri case sensitive blah', + ) + self.assertEqual( + self.get(url='/BlaH')['status'], + 404, + 'match uri case sensitive BlaH', + ) + self.assertEqual( + self.get(url='/BLAH')['status'], + 200, + 'match uri case sensitive BLAH', + ) def test_routes_match_uri_normalize(self): - self.assertIn('success', self.conf([{ - "match": { "uri": "/blah" }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'match uri normalize configure') - - self.assertEqual(self.get(url='/%62%6c%61%68')['status'], 200, - 'match uri normalize') + self.assertIn( + 'success', + self.conf( + [ + { + "match": {"uri": "/blah"}, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'match uri normalize configure', + ) + + self.assertEqual( + self.get(url='/%62%6c%61%68')['status'], 200, 'match uri normalize' + ) def test_routes_match_rules(self): - self.assertIn('success', self.conf([{ - "match": { - "method": "GET", - "host": "localhost", - "uri": "/" - }, - "action": { "pass": "applications/empty" } - }], 'routes'), 'routes match rules configure') + self.assertIn( + 'success', + self.conf( + [ + { + "match": { + "method": "GET", + "host": "localhost", + "uri": "/", + }, + "action": {"pass": "applications/empty"}, + } + ], + 'routes', + ), + 'routes match rules configure', + ) self.assertEqual(self.get()['status'], 200, 'routes match rules') def test_routes_loop(self): - self.assertIn('success', self.conf([{ - "match": { "uri": "/" }, - "action": { "pass": "routes" } - }], 'routes'), 'routes loop configure') + self.assertIn( + 'success', + self.conf( + [{"match": {"uri": "/"}, "action": {"pass": "routes"}}], + 'routes', + ), + 'routes loop configure', + ) self.assertEqual(self.get()['status'], 500, 'routes loop') + if __name__ == '__main__': TestUnitRouting.main() diff --git a/test/test_ruby_application.py b/test/test_ruby_application.py index 262fc497..0f2f6185 100644 --- a/test/test_ruby_application.py +++ b/test/test_ruby_application.py @@ -1,8 +1,8 @@ import unittest import unit -class TestUnitRubyApplication(unit.TestUnitApplicationRuby): +class TestUnitRubyApplication(unit.TestUnitApplicationRuby): def setUpClass(): unit.TestUnit().check_modules('ruby') @@ -11,43 +11,56 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): body = 'Test body string.' - resp = self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': 'blah', - 'Connection': 'close' - }, body=body) + resp = self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': 'blah', + 'Connection': 'close', + }, + body=body, + ) self.assertEqual(resp['status'], 200, 'status') headers = resp['headers'] header_server = headers.pop('Server') self.assertRegex(header_server, r'Unit/[\d\.]+', 'server header') - self.assertEqual(headers.pop('Server-Software'), header_server, - 'server software header') + self.assertEqual( + headers.pop('Server-Software'), + header_server, + 'server software header', + ) date = headers.pop('Date') self.assertEqual(date[-4:], ' GMT', 'date header timezone') - self.assertLess(abs(self.date_to_sec_epoch(date) - self.sec_epoch()), 5, - 'date header') - - self.assertDictEqual(headers, { - 'Connection': 'close', - 'Content-Length': str(len(body)), - 'Content-Type': 'text/html', - 'Request-Method': 'POST', - 'Request-Uri': '/', - 'Http-Host': 'localhost', - 'Server-Protocol': 'HTTP/1.1', - 'Custom-Header': 'blah', - 'Rack-Version': '13', - 'Rack-Url-Scheme': 'http', - 'Rack-Multithread': 'false', - 'Rack-Multiprocess': 'true', - 'Rack-Run-Once': 'false', - 'Rack-Hijack-Q': 'false', - 'Rack-Hijack': '', - 'Rack-Hijack-IO': '' - }, 'headers') + self.assertLess( + abs(self.date_to_sec_epoch(date) - self.sec_epoch()), + 5, + 'date header', + ) + + self.assertDictEqual( + headers, + { + 'Connection': 'close', + 'Content-Length': str(len(body)), + 'Content-Type': 'text/html', + 'Request-Method': 'POST', + 'Request-Uri': '/', + 'Http-Host': 'localhost', + 'Server-Protocol': 'HTTP/1.1', + 'Custom-Header': 'blah', + 'Rack-Version': '13', + 'Rack-Url-Scheme': 'http', + 'Rack-Multithread': 'false', + 'Rack-Multiprocess': 'true', + 'Rack-Run-Once': 'false', + 'Rack-Hijack-Q': 'false', + 'Rack-Hijack': '', + 'Rack-Hijack-IO': '', + }, + 'headers', + ) self.assertEqual(resp['body'], body, 'body') def test_ruby_application_query_string(self): @@ -55,8 +68,11 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): resp = self.get(url='/?var1=val1&var2=val2') - self.assertEqual(resp['headers']['Query-String'], 'var1=val1&var2=val2', - 'Query-String header') + self.assertEqual( + resp['headers']['Query-String'], + 'var1=val1&var2=val2', + 'Query-String header', + ) def test_ruby_application_query_string_empty(self): self.load('query_string') @@ -64,8 +80,9 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): resp = self.get(url='/?') self.assertEqual(resp['status'], 200, 'query string empty status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string empty') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string empty' + ) @unittest.expectedFailure def test_ruby_application_query_string_absent(self): @@ -74,15 +91,17 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): resp = self.get() self.assertEqual(resp['status'], 200, 'query string absent status') - self.assertEqual(resp['headers']['Query-String'], '', - 'query string absent') + self.assertEqual( + resp['headers']['Query-String'], '', 'query string absent' + ) @unittest.expectedFailure def test_ruby_application_server_port(self): self.load('server_port') - self.assertEqual(self.get()['headers']['Server-Port'], '7080', - 'Server-Port header') + self.assertEqual( + self.get()['headers']['Server-Port'], '7080', 'Server-Port header' + ) def test_ruby_application_status_int(self): self.load('status_int') @@ -97,20 +116,29 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): def test_ruby_application_input_read_parts(self): self.load('input_read_parts') - self.assertEqual(self.post(body='0123456789')['body'], '012345678', - 'input read parts') + self.assertEqual( + self.post(body='0123456789')['body'], + '012345678', + 'input read parts', + ) def test_ruby_application_input_read_buffer(self): self.load('input_read_buffer') - self.assertEqual(self.post(body='0123456789')['body'], '0123456789', - 'input read buffer') + self.assertEqual( + self.post(body='0123456789')['body'], + '0123456789', + 'input read buffer', + ) def test_ruby_application_input_read_buffer_not_empty(self): self.load('input_read_buffer_not_empty') - self.assertEqual(self.post(body='0123456789')['body'], '0123456789', - 'input read buffer not empty') + self.assertEqual( + self.post(body='0123456789')['body'], + '0123456789', + 'input read buffer not empty', + ) def test_ruby_application_input_gets(self): self.load('input_gets') @@ -122,8 +150,9 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): def test_ruby_application_input_gets_2(self): self.load('input_gets') - self.assertEqual(self.post(body='01234\n56789\n')['body'], '01234\n', - 'input gets 2') + self.assertEqual( + self.post(body='01234\n56789\n')['body'], '01234\n', 'input gets 2' + ) def test_ruby_application_input_gets_all(self): self.load('input_gets_all') @@ -149,12 +178,14 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): @unittest.expectedFailure def test_ruby_application_syntax_error(self): - self.skip_alerts.extend([ - r'Failed to parse rack script', - r'syntax error', - r'new_from_string', - r'parse_file' - ]) + self.skip_alerts.extend( + [ + r'Failed to parse rack script', + r'syntax error', + r'new_from_string', + r'parse_file', + ] + ) self.load('syntax_error') self.assertEqual(self.get()['status'], 500, 'syntax error') @@ -168,7 +199,8 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): self.assertIsNotNone( self.search_in_log(r'\[error\].+Error in application'), - 'errors puts') + 'errors puts', + ) def test_ruby_application_errors_puts_int(self): self.load('errors_puts_int') @@ -178,8 +210,8 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): self.stop() self.assertIsNotNone( - self.search_in_log(r'\[error\].+1234567890'), - 'errors puts int') + self.search_in_log(r'\[error\].+1234567890'), 'errors puts int' + ) def test_ruby_application_errors_write(self): self.load('errors_write') @@ -190,13 +222,13 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): self.assertIsNotNone( self.search_in_log(r'\[error\].+Error in application'), - 'errors write') + 'errors write', + ) def test_ruby_application_errors_write_to_s_custom(self): self.load('errors_write_to_s_custom') - self.assertEqual(self.get()['status'], 200, - 'errors write to_s custom') + self.assertEqual(self.get()['status'], 200, 'errors write to_s custom') def test_ruby_application_errors_write_int(self): self.load('errors_write_int') @@ -206,38 +238,40 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): self.stop() self.assertIsNotNone( - self.search_in_log(r'\[error\].+1234567890'), - 'errors write int') + self.search_in_log(r'\[error\].+1234567890'), 'errors write int' + ) def test_ruby_application_at_exit(self): self.load('at_exit') self.get() - self.conf({ - "listeners": {}, - "applications": {} - }) + self.conf({"listeners": {}, "applications": {}}) self.stop() self.assertIsNotNone( - self.search_in_log(r'\[error\].+At exit called\.'), 'at exit') + self.search_in_log(r'\[error\].+At exit called\.'), 'at exit' + ) def test_ruby_application_header_custom(self): self.load('header_custom') resp = self.post(body="\ntc=one,two\ntc=three,four,\n\n") - self.assertEqual(resp['headers']['Custom-Header'], - ['', 'tc=one,two', 'tc=three,four,', '', ''], 'header custom') + self.assertEqual( + resp['headers']['Custom-Header'], + ['', 'tc=one,two', 'tc=three,four,', '', ''], + 'header custom', + ) @unittest.expectedFailure def test_ruby_application_header_custom_non_printable(self): self.load('header_custom') - self.assertEqual(self.post(body='\b')['status'], 500, - 'header custom non printable') + self.assertEqual( + self.post(body='\b')['status'], 500, 'header custom non printable' + ) def test_ruby_application_header_status(self): self.load('header_status') @@ -277,7 +311,8 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): self.assertIsNotNone( self.search_in_log(r'\[error\].+Failed to run ruby script'), - 'body each error') + 'body each error', + ) def test_ruby_application_body_file(self): self.load('body_file') @@ -287,21 +322,30 @@ class TestUnitRubyApplication(unit.TestUnitApplicationRuby): def test_ruby_keepalive_body(self): self.load('mirror') - (resp, sock) = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789' * 500) + (resp, sock) = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789' * 500, + ) self.assertEqual(resp['body'], '0123456789' * 500, 'keep-alive 1') - resp = self.post(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keep-alive 2') + if __name__ == '__main__': TestUnitRubyApplication.main() diff --git a/test/test_settings.py b/test/test_settings.py index 13bfad49..85bbfb44 100644 --- a/test/test_settings.py +++ b/test/test_settings.py @@ -3,45 +3,73 @@ import socket import unittest import unit -class TestUnitSettings(unit.TestUnitApplicationPython): +class TestUnitSettings(unit.TestUnitApplicationPython): def setUpClass(): unit.TestUnit().check_modules('python') def test_settings_header_read_timeout(self): self.load('empty') - self.conf({'http': { 'header_read_timeout': 2 }}, 'settings') + self.conf({'http': {'header_read_timeout': 2}}, 'settings') - (resp, sock) = self.http(b"""GET / HTTP/1.1 -""", start=True, read_timeout=1, raw=True) + (resp, sock) = self.http( + b"""GET / HTTP/1.1 +""", + start=True, + read_timeout=1, + raw=True, + ) time.sleep(3) - resp = self.http(b"""Host: localhost + resp = self.http( + b"""Host: localhost Connection: close -""", sock=sock, raw=True) +""", + sock=sock, + raw=True, + ) self.assertEqual(resp['status'], 408, 'status header read timeout') def test_settings_header_read_timeout_update(self): self.load('empty') - self.conf({'http': { 'header_read_timeout': 4 }}, 'settings') + self.conf({'http': {'header_read_timeout': 4}}, 'settings') - (resp, sock) = self.http(b"""GET / HTTP/1.1 -""", start=True, read_timeout=1, raw=True, no_recv=True) + (resp, sock) = self.http( + b"""GET / HTTP/1.1 +""", + start=True, + read_timeout=1, + raw=True, + no_recv=True, + ) time.sleep(2) - (resp, sock) = self.http(b"""Host: localhost -""", start=True, sock=sock, read_timeout=1, raw=True, no_recv=True) + (resp, sock) = self.http( + b"""Host: localhost +""", + start=True, + sock=sock, + read_timeout=1, + raw=True, + no_recv=True, + ) time.sleep(2) - (resp, sock) = self.http(b"""X-Blah: blah -""", start=True, sock=sock, read_timeout=1, raw=True) + (resp, sock) = self.http( + b"""X-Blah: blah +""", + start=True, + sock=sock, + read_timeout=1, + raw=True, + ) if len(resp) != 0: sock.close() @@ -49,24 +77,35 @@ Connection: close else: time.sleep(2) - resp = self.http(b"""Connection: close + resp = self.http( + b"""Connection: close -""", sock=sock, raw=True) +""", + sock=sock, + raw=True, + ) - self.assertEqual(resp['status'], 408, - 'status header read timeout update') + self.assertEqual( + resp['status'], 408, 'status header read timeout update' + ) def test_settings_body_read_timeout(self): self.load('empty') - self.conf({'http': { 'body_read_timeout': 2 }}, 'settings') + self.conf({'http': {'body_read_timeout': 2}}, 'settings') - (resp, sock) = self.http(b"""POST / HTTP/1.1 + (resp, sock) = self.http( + b"""POST / HTTP/1.1 Host: localhost Content-Length: 10 Connection: close -""", start=True, raw_resp=True, read_timeout=1, raw=True) +""", + start=True, + raw_resp=True, + read_timeout=1, + raw=True, + ) time.sleep(3) @@ -77,37 +116,46 @@ Connection: close def test_settings_body_read_timeout_update(self): self.load('empty') - self.conf({'http': { 'body_read_timeout': 4 }}, 'settings') + self.conf({'http': {'body_read_timeout': 4}}, 'settings') - (resp, sock) = self.http(b"""POST / HTTP/1.1 + (resp, sock) = self.http( + b"""POST / HTTP/1.1 Host: localhost Content-Length: 10 Connection: close -""", start=True, read_timeout=1, raw=True) +""", + start=True, + read_timeout=1, + raw=True, + ) time.sleep(2) - (resp, sock) = self.http(b"""012""", start=True, sock=sock, - read_timeout=1, raw=True) + (resp, sock) = self.http( + b"""012""", start=True, sock=sock, read_timeout=1, raw=True + ) time.sleep(2) - (resp, sock) = self.http(b"""345""", start=True, sock=sock, - read_timeout=1, raw=True) + (resp, sock) = self.http( + b"""345""", start=True, sock=sock, read_timeout=1, raw=True + ) time.sleep(2) resp = self.http(b"""6789""", sock=sock, raw=True) - self.assertEqual(resp['status'], 200, 'status body read timeout update') + self.assertEqual( + resp['status'], 200, 'status body read timeout update' + ) def test_settings_send_timeout(self): self.load('mirror') data_len = 1048576 - self.conf({'http': { 'send_timeout': 1 }}, 'settings') + self.conf({'http': {'send_timeout': 1}}, 'settings') addr = self.testdir + '/sock' @@ -122,7 +170,9 @@ Content-Type: text/html Content-Length: %d Connection: close -""" % data_len + ('X' * data_len) +""" % data_len + ( + 'X' * data_len + ) sock.sendall(req.encode()) @@ -140,35 +190,40 @@ Connection: close def test_settings_idle_timeout(self): self.load('empty') - self.conf({'http': { 'idle_timeout': 2 }}, 'settings') + self.conf({'http': {'idle_timeout': 2}}, 'settings') - (resp, sock) = self.get(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive' - }, start=True, read_timeout=1) + (resp, sock) = self.get( + headers={'Host': 'localhost', 'Connection': 'keep-alive'}, + start=True, + read_timeout=1, + ) time.sleep(3) - resp = self.get(headers={ - 'Host': 'localhost', - 'Connection': 'close' - }, sock=sock) + resp = self.get( + headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock + ) self.assertEqual(resp['status'], 408, 'status idle timeout') def test_settings_max_body_size(self): self.load('empty') - self.conf({'http': { 'max_body_size': 5 }}, 'settings') + self.conf({'http': {'max_body_size': 5}}, 'settings') self.assertEqual(self.post(body='01234')['status'], 200, 'status size') - self.assertEqual(self.post(body='012345')['status'], 413, - 'status size max') + self.assertEqual( + self.post(body='012345')['status'], 413, 'status size max' + ) @unittest.expectedFailure def test_settings_negative_value(self): - self.assertIn('error', self.conf({'http': { 'max_body_size': -1 }}, - 'settings'), 'settings negative value') + self.assertIn( + 'error', + self.conf({'http': {'max_body_size': -1}}, 'settings'), + 'settings negative value', + ) + if __name__ == '__main__': TestUnitSettings.main() diff --git a/test/test_tls.py b/test/test_tls.py index d9b2e512..8b112f4e 100644 --- a/test/test_tls.py +++ b/test/test_tls.py @@ -5,8 +5,8 @@ import subprocess import unittest import unit -class TestUnitTLS(unit.TestUnitApplicationTLS): +class TestUnitTLS(unit.TestUnitApplicationTLS): def setUpClass(): unit.TestUnit().check_modules('python', 'openssl') @@ -26,17 +26,13 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): return self.date_to_sec_epoch(date, '%b %d %H:%M:%S %Y %Z') def add_tls(self, application='empty', cert='default', port=7080): - self.conf({ - "application": application, - "tls": { - "certificate": cert - } - }, 'listeners/*:' + str(port)) + self.conf( + {"application": application, "tls": {"certificate": cert}}, + 'listeners/*:' + str(port), + ) def remove_tls(self, application='empty', port=7080): - self.conf({ - "application": application - }, 'listeners/*:' + str(port)) + self.conf({"application": application}, 'listeners/*:' + str(port)) def test_tls_listener_option_add(self): self.load('empty') @@ -65,8 +61,11 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.certificate() - self.assertIn('success', self.conf_delete('/certificates/default'), - 'remove certificate') + self.assertIn( + 'success', + self.conf_delete('/certificates/default'), + 'remove certificate', + ) def test_tls_certificate_remove_used(self): self.load('empty') @@ -75,8 +74,11 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.add_tls() - self.assertIn('error', self.conf_delete('/certificates/default'), - 'remove certificate') + self.assertIn( + 'error', + self.conf_delete('/certificates/default'), + 'remove certificate', + ) def test_tls_certificate_remove_nonexisting(self): self.load('empty') @@ -85,8 +87,11 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.add_tls() - self.assertIn('error', self.conf_delete('/certificates/blah'), - 'remove nonexistings certificate') + self.assertIn( + 'error', + self.conf_delete('/certificates/blah'), + 'remove nonexistings certificate', + ) @unittest.expectedFailure def test_tls_certificate_update(self): @@ -100,8 +105,9 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.certificate() - self.assertNotEqual(cert_old, self.get_server_certificate(), - 'update certificate') + self.assertNotEqual( + cert_old, self.get_server_certificate(), 'update certificate' + ) @unittest.expectedFailure def test_tls_certificate_key_incorrect(self): @@ -110,8 +116,9 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.certificate('first', False) self.certificate('second', False) - self.assertIn('error', self.certificate_load('first', 'second'), - 'key incorrect') + self.assertIn( + 'error', self.certificate_load('first', 'second'), 'key incorrect' + ) def test_tls_certificate_change(self): self.load('empty') @@ -125,33 +132,53 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): self.add_tls(cert='new') - self.assertNotEqual(cert_old, self.get_server_certificate(), - 'change certificate') + self.assertNotEqual( + cert_old, self.get_server_certificate(), 'change certificate' + ) def test_tls_certificate_key_rsa(self): self.load('empty') self.certificate() - self.assertEqual(self.conf_get('/certificates/default/key'), - 'RSA (1024 bits)', 'certificate key rsa') + self.assertEqual( + self.conf_get('/certificates/default/key'), + 'RSA (1024 bits)', + 'certificate key rsa', + ) def test_tls_certificate_key_ec(self): self.load('empty') - subprocess.call(['openssl', 'ecparam', '-noout', '-genkey', - '-out', self.testdir + '/ec.key', - '-name', 'prime256v1']) - - subprocess.call(['openssl', 'req', '-x509', '-new', - '-config', self.testdir + '/openssl.conf', - '-key', self.testdir + '/ec.key', '-subj', '/CN=ec/', - '-out', self.testdir + '/ec.crt']) + subprocess.call( + [ + 'openssl', + 'ecparam', + '-noout', + '-genkey', + '-out', self.testdir + '/ec.key', + '-name', 'prime256v1', + ] + ) + + subprocess.call( + [ + 'openssl', + 'req', + '-x509', + '-new', + '-subj', '/CN=ec/', + '-config', self.testdir + '/openssl.conf', + '-key', self.testdir + '/ec.key', + '-out', self.testdir + '/ec.crt', + ] + ) self.certificate_load('ec') - self.assertEqual(self.conf_get('/certificates/ec/key'), 'ECDH', - 'certificate key ec') + self.assertEqual( + self.conf_get('/certificates/ec/key'), 'ECDH', 'certificate key ec' + ) def test_tls_certificate_chain_options(self): self.load('empty') @@ -164,36 +191,64 @@ class TestUnitTLS(unit.TestUnitApplicationTLS): cert = chain[0] - self.assertEqual(cert['subject']['common_name'], 'default', - 'certificate subject common name') - self.assertEqual(cert['issuer']['common_name'], 'default', - 'certificate issuer common name') - - self.assertLess(abs(self.sec_epoch() - - self.openssl_date_to_sec_epoch(cert['validity']['since'])), 5, - 'certificate validity since') self.assertEqual( - self.openssl_date_to_sec_epoch(cert['validity']['until']) - - self.openssl_date_to_sec_epoch(cert['validity']['since']), 2592000, - 'certificate validity until') + cert['subject']['common_name'], + 'default', + 'certificate subject common name', + ) + self.assertEqual( + cert['issuer']['common_name'], + 'default', + 'certificate issuer common name', + ) + + self.assertLess( + abs( + self.sec_epoch() + - self.openssl_date_to_sec_epoch(cert['validity']['since']) + ), + 5, + 'certificate validity since', + ) + self.assertEqual( + self.openssl_date_to_sec_epoch(cert['validity']['until']) + - self.openssl_date_to_sec_epoch(cert['validity']['since']), + 2592000, + 'certificate validity until', + ) def test_tls_certificate_chain(self): self.load('empty') self.certificate('root', False) - subprocess.call(['openssl', 'req', '-new', '-config', - self.testdir + '/openssl.conf', '-subj', '/CN=int/', - '-out', self.testdir + '/int.csr', - '-keyout', self.testdir + '/int.key']) - - subprocess.call(['openssl', 'req', '-new', '-config', - self.testdir + '/openssl.conf', '-subj', '/CN=end/', - '-out', self.testdir + '/end.csr', - '-keyout', self.testdir + '/end.key']) + subprocess.call( + [ + 'openssl', + 'req', + '-new', + '-subj', '/CN=int/', + '-config', self.testdir + '/openssl.conf', + '-out', self.testdir + '/int.csr', + '-keyout', self.testdir + '/int.key', + ] + ) + + subprocess.call( + [ + 'openssl', + 'req', + '-new', + '-subj', '/CN=end/', + '-config', self.testdir + '/openssl.conf', + '-out', self.testdir + '/end.csr', + '-keyout', self.testdir + '/end.key', + ] + ) with open(self.testdir + '/ca.conf', 'w') as f: - f.write("""[ ca ] + f.write( + """[ ca ] default_ca = myca [ myca ] @@ -209,11 +264,13 @@ x509_extensions = myca_extensions commonName = supplied [ myca_extensions ] -basicConstraints = critical,CA:TRUE""" % { - 'dir': self.testdir, - 'database': self.testdir + '/certindex', - 'certserial': self.testdir + '/certserial' - }) +basicConstraints = critical,CA:TRUE""" + % { + 'dir': self.testdir, + 'database': self.testdir + '/certindex', + 'certserial': self.testdir + '/certserial', + } + ) with open(self.testdir + '/certserial', 'w') as f: f.write('1000') @@ -221,26 +278,42 @@ basicConstraints = critical,CA:TRUE""" % { with open(self.testdir + '/certindex', 'w') as f: f.write('') - subprocess.call(['openssl', 'ca', '-batch', - '-config', self.testdir + '/ca.conf', - '-keyfile', self.testdir + '/root.key', - '-cert', self.testdir + '/root.crt', - '-subj', '/CN=int/', - '-in', self.testdir + '/int.csr', - '-out', self.testdir + '/int.crt']) - - subprocess.call(['openssl', 'ca', '-batch', - '-config', self.testdir + '/ca.conf', - '-keyfile', self.testdir + '/int.key', - '-cert', self.testdir + '/int.crt', - '-subj', '/CN=end/', - '-in', self.testdir + '/end.csr', - '-out', self.testdir + '/end.crt']) - - with open(self.testdir + '/end-int.crt', 'wb') as crt, \ - open(self.testdir + '/end.crt', 'rb') as end, \ - open(self.testdir + '/int.crt', 'rb') as int: - crt.write(end.read() + int.read()) + subprocess.call( + [ + 'openssl', + 'ca', + '-batch', + '-subj', '/CN=int/', + '-config', self.testdir + '/ca.conf', + '-keyfile', self.testdir + '/root.key', + '-cert', self.testdir + '/root.crt', + '-in', self.testdir + '/int.csr', + '-out', self.testdir + '/int.crt', + ] + ) + + subprocess.call( + [ + 'openssl', + 'ca', + '-batch', + '-subj', '/CN=end/', + '-config', self.testdir + '/ca.conf', + '-keyfile', self.testdir + '/int.key', + '-cert', self.testdir + '/int.crt', + '-in', self.testdir + '/end.csr', + '-out', self.testdir + '/end.crt', + ] + ) + + crt_path = self.testdir + '/end-int.crt' + end_path = self.testdir + '/end.crt' + int_path = self.testdir + '/int.crt' + + with open(crt_path, 'wb') as crt, \ + open(end_path, 'rb') as end, \ + open(int_path, 'rb') as int: + crt.write(end.read() + int.read()) self.context = ssl.create_default_context() self.context.check_hostname = False @@ -249,15 +322,24 @@ basicConstraints = critical,CA:TRUE""" % { # incomplete chain - self.assertIn('success', self.certificate_load('end', 'end'), - 'certificate chain end upload') + self.assertIn( + 'success', + self.certificate_load('end', 'end'), + 'certificate chain end upload', + ) chain = self.conf_get('/certificates/end/chain') self.assertEqual(len(chain), 1, 'certificate chain end length') - self.assertEqual(chain[0]['subject']['common_name'], 'end', - 'certificate chain end subject common name') - self.assertEqual(chain[0]['issuer']['common_name'], 'int', - 'certificate chain end issuer common name') + self.assertEqual( + chain[0]['subject']['common_name'], + 'end', + 'certificate chain end subject common name', + ) + self.assertEqual( + chain[0]['issuer']['common_name'], + 'int', + 'certificate chain end issuer common name', + ) self.add_tls(cert='end') @@ -270,41 +352,69 @@ basicConstraints = critical,CA:TRUE""" % { # intermediate - self.assertIn('success', self.certificate_load('int', 'int'), - 'certificate chain int upload') + self.assertIn( + 'success', + self.certificate_load('int', 'int'), + 'certificate chain int upload', + ) chain = self.conf_get('/certificates/int/chain') self.assertEqual(len(chain), 1, 'certificate chain int length') - self.assertEqual(chain[0]['subject']['common_name'], 'int', - 'certificate chain int subject common name') - self.assertEqual(chain[0]['issuer']['common_name'], 'root', - 'certificate chain int issuer common name') + self.assertEqual( + chain[0]['subject']['common_name'], + 'int', + 'certificate chain int subject common name', + ) + self.assertEqual( + chain[0]['issuer']['common_name'], + 'root', + 'certificate chain int issuer common name', + ) self.add_tls(cert='int') - self.assertEqual(self.get_ssl()['status'], 200, - 'certificate chain intermediate') + self.assertEqual( + self.get_ssl()['status'], 200, 'certificate chain intermediate' + ) # intermediate server - self.assertIn('success', self.certificate_load('end-int', 'end'), - 'certificate chain end-int upload') + self.assertIn( + 'success', + self.certificate_load('end-int', 'end'), + 'certificate chain end-int upload', + ) chain = self.conf_get('/certificates/end-int/chain') self.assertEqual(len(chain), 2, 'certificate chain end-int length') - self.assertEqual(chain[0]['subject']['common_name'], 'end', - 'certificate chain end-int int subject common name') - self.assertEqual(chain[0]['issuer']['common_name'], 'int', - 'certificate chain end-int int issuer common name') - self.assertEqual(chain[1]['subject']['common_name'], 'int', - 'certificate chain end-int end subject common name') - self.assertEqual(chain[1]['issuer']['common_name'], 'root', - 'certificate chain end-int end issuer common name') + self.assertEqual( + chain[0]['subject']['common_name'], + 'end', + 'certificate chain end-int int subject common name', + ) + self.assertEqual( + chain[0]['issuer']['common_name'], + 'int', + 'certificate chain end-int int issuer common name', + ) + self.assertEqual( + chain[1]['subject']['common_name'], + 'int', + 'certificate chain end-int end subject common name', + ) + self.assertEqual( + chain[1]['issuer']['common_name'], + 'root', + 'certificate chain end-int end issuer common name', + ) self.add_tls(cert='end-int') - self.assertEqual(self.get_ssl()['status'], 200, - 'certificate chain intermediate server') + self.assertEqual( + self.get_ssl()['status'], + 200, + 'certificate chain intermediate server', + ) @unittest.expectedFailure def test_tls_reconfigure(self): @@ -312,19 +422,21 @@ basicConstraints = critical,CA:TRUE""" % { self.certificate() - (resp, sock) = self.get(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive' - }, start=True) + (resp, sock) = self.get( + headers={'Host': 'localhost', 'Connection': 'keep-alive'}, + start=True, + ) self.assertEqual(resp['status'], 200, 'initial status') self.add_tls() - self.assertEqual(self.get(sock=sock)['status'], 200, - 'reconfigure status') - self.assertEqual(self.get_ssl()['status'], 200, - 'reconfigure tls status') + self.assertEqual( + self.get(sock=sock)['status'], 200, 'reconfigure status' + ) + self.assertEqual( + self.get_ssl()['status'], 200, 'reconfigure tls status' + ) def test_tls_keepalive(self): self.load('mirror') @@ -333,19 +445,27 @@ basicConstraints = critical,CA:TRUE""" % { self.add_tls(application='mirror') - (resp, sock) = self.post_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789') + (resp, sock) = self.post_ssl( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keepalive 1') - resp = self.post_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post_ssl( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['body'], '0123456789', 'keepalive 2') @@ -357,21 +477,18 @@ basicConstraints = critical,CA:TRUE""" % { self.add_tls() - (resp, sock) = self.get_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive' - }, start=True) + (resp, sock) = self.get_ssl( + headers={'Host': 'localhost', 'Connection': 'keep-alive'}, + start=True, + ) - self.conf({ - "application": "empty" - }, 'listeners/*:7080') + self.conf({"application": "empty"}, 'listeners/*:7080') self.conf_delete('/certificates/default') try: - resp = self.get_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'close' - }, sock=sock) + resp = self.get_ssl( + headers={'Host': 'localhost', 'Connection': 'close'}, sock=sock + ) except: resp = None @@ -383,8 +500,11 @@ basicConstraints = critical,CA:TRUE""" % { self.certificate() - self.assertIn('success', self.conf_delete('/certificates'), - 'remove all certificates') + self.assertIn( + 'success', + self.conf_delete('/certificates'), + 'remove all certificates', + ) def test_tls_application_respawn(self): self.skip_alerts.append(r'process \d+ exited on signal 9') @@ -396,48 +516,73 @@ basicConstraints = critical,CA:TRUE""" % { self.add_tls(application='mirror') - (resp, sock) = self.post_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'keep-alive', - 'Content-Type': 'text/html' - }, start=True, body='0123456789') + (resp, sock) = self.post_ssl( + headers={ + 'Host': 'localhost', + 'Connection': 'keep-alive', + 'Content-Type': 'text/html', + }, + start=True, + body='0123456789', + ) app_id = self.findall(r'(\d+)#\d+ "mirror" application started')[0] subprocess.call(['kill', '-9', app_id]) - self.wait_for_record(re.compile(' (?!' + app_id + - '#)(\d+)#\d+ "mirror" application started')) + self.wait_for_record( + re.compile( + ' (?!' + app_id + '#)(\d+)#\d+ "mirror" application started' + ) + ) - resp = self.post_ssl(headers={ - 'Host': 'localhost', - 'Connection': 'close', - 'Content-Type': 'text/html' - }, sock=sock, body='0123456789') + resp = self.post_ssl( + headers={ + 'Host': 'localhost', + 'Connection': 'close', + 'Content-Type': 'text/html', + }, + sock=sock, + body='0123456789', + ) self.assertEqual(resp['status'], 200, 'application respawn status') - self.assertEqual(resp['body'], '0123456789', 'application respawn body') + self.assertEqual( + resp['body'], '0123456789', 'application respawn body' + ) def test_tls_url_scheme(self): self.load('variables') - self.assertEqual(self.post(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': '', - 'Connection': 'close' - })['headers']['Wsgi-Url-Scheme'], 'http', 'url scheme http') + self.assertEqual( + self.post( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': '', + 'Connection': 'close', + } + )['headers']['Wsgi-Url-Scheme'], + 'http', + 'url scheme http', + ) self.certificate() self.add_tls(application='variables') - self.assertEqual(self.post_ssl(headers={ - 'Host': 'localhost', - 'Content-Type': 'text/html', - 'Custom-Header': '', - 'Connection': 'close' - })['headers']['Wsgi-Url-Scheme'], 'https', 'url scheme https') + self.assertEqual( + self.post_ssl( + headers={ + 'Host': 'localhost', + 'Content-Type': 'text/html', + 'Custom-Header': '', + 'Connection': 'close', + } + )['headers']['Wsgi-Url-Scheme'], + 'https', + 'url scheme https', + ) if __name__ == '__main__': TestUnitTLS.main() diff --git a/test/unit.py b/test/unit.py index 6cca7f48..4a613d1b 100644 --- a/test/unit.py +++ b/test/unit.py @@ -14,9 +14,12 @@ import unittest import subprocess from multiprocessing import Process + class TestUnit(unittest.TestCase): - pardir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) + pardir = os.path.abspath( + os.path.join(os.path.dirname(__file__), os.pardir) + ) architecture = platform.architecture()[0] maxDiff = None @@ -61,16 +64,19 @@ class TestUnit(unittest.TestCase): result = self.defaultTestResult() self._feedErrorsToResult(result, self._outcome.errors) else: - result = getattr(self, '_outcomeForDoCleanups', - self._resultForDoCleanups) + result = getattr( + self, '_outcomeForDoCleanups', self._resultForDoCleanups + ) - success = not list2reason(result.errors) \ - and not list2reason(result.failures) + success = not list2reason(result.errors) and not list2reason( + result.failures + ) # check unit.log for alerts - with open(self.testdir + '/unit.log', 'r', encoding='utf-8', - errors='ignore') as f: + unit_log = self.testdir + '/unit.log' + + with open(unit_log, 'r', encoding='utf-8', errors='ignore') as f: self._check_alerts(f.read()) # remove unit.log @@ -107,9 +113,16 @@ class TestUnit(unittest.TestCase): env['GOPATH'] = self.pardir + '/go' try: - process = subprocess.Popen(['go', 'build', '-o', - self.testdir + '/go/check_module', - current_dir + '/go/empty/app.go'], env=env) + process = subprocess.Popen( + [ + 'go', + 'build', + '-o', + self.testdir + '/go/check_module', + current_dir + '/go/empty/app.go', + ], + env=env, + ) process.communicate() m = module if process.returncode == 0 else None @@ -127,9 +140,10 @@ class TestUnit(unittest.TestCase): try: subprocess.check_output(['which', 'openssl']) - output = subprocess.check_output([ - self.pardir + '/build/unitd', '--version'], - stderr=subprocess.STDOUT) + output = subprocess.check_output( + [self.pardir + '/build/unitd', '--version'], + stderr=subprocess.STDOUT, + ) m = re.search('--openssl', output.decode()) @@ -162,25 +176,35 @@ class TestUnit(unittest.TestCase): print() def _run_unit(): - subprocess.call([self.pardir + '/build/unitd', - '--no-daemon', - '--modules', self.pardir + '/build', - '--state', self.testdir + '/state', - '--pid', self.testdir + '/unit.pid', - '--log', self.testdir + '/unit.log', - '--control', 'unix:' + self.testdir + '/control.unit.sock']) + subprocess.call( + [ + self.pardir + '/build/unitd', + '--no-daemon', + '--modules', self.pardir + '/build', + '--state', self.testdir + '/state', + '--pid', self.testdir + '/unit.pid', + '--log', self.testdir + '/unit.log', + '--control', 'unix:' + self.testdir + '/control.unit.sock', + ] + ) self._p = Process(target=_run_unit) self._p.start() - if not self.waitforfiles(self.testdir + '/unit.pid', - self.testdir + '/unit.log', self.testdir + '/control.unit.sock'): + if not self.waitforfiles( + self.testdir + '/unit.pid', + self.testdir + '/unit.log', + self.testdir + '/control.unit.sock', + ): exit("Could not start unit") self._started = True - self.skip_alerts = [r'read signalfd\(4\) failed', r'sendmsg.+failed', - r'recvmsg.+failed'] + self.skip_alerts = [ + r'read signalfd\(4\) failed', + r'sendmsg.+failed', + r'recvmsg.+failed', + ] self.skip_sanitizer = False def _stop(self): @@ -248,8 +272,8 @@ class TestUnit(unittest.TestCase): for f in files: if not os.path.exists(f): - wait = True - break + wait = True + break if wait: time.sleep(0.1) @@ -264,10 +288,20 @@ class TestUnit(unittest.TestCase): def _parse_args(): parser = argparse.ArgumentParser(add_help=False) - parser.add_argument('-d', '--detailed', dest='detailed', - action='store_true', help='Detailed output for tests') - parser.add_argument('-l', '--log', dest='save_log', - action='store_true', help='Save unit.log after the test execution') + parser.add_argument( + '-d', + '--detailed', + dest='detailed', + action='store_true', + help='Detailed output for tests', + ) + parser.add_argument( + '-l', + '--log', + dest='save_log', + action='store_true', + help='Save unit.log after the test execution', + ) return parser.parse_known_args() @@ -279,18 +313,21 @@ class TestUnit(unittest.TestCase): def _print_path_to_log(self): print('Path to unit.log:\n' + self.testdir + '/unit.log') -class TestUnitHTTP(TestUnit): +class TestUnitHTTP(TestUnit): def http(self, start_str, **kwargs): - sock_type = 'ipv4' if 'sock_type' not in kwargs else kwargs['sock_type'] + sock_type = ( + 'ipv4' if 'sock_type' not in kwargs else kwargs['sock_type'] + ) port = 7080 if 'port' not in kwargs else kwargs['port'] url = '/' if 'url' not in kwargs else kwargs['url'] http = 'HTTP/1.0' if 'http_10' in kwargs else 'HTTP/1.1' - headers = ({ - 'Host': 'localhost', - 'Connection': 'close' - } if 'headers' not in kwargs else kwargs['headers']) + headers = ( + {'Host': 'localhost', 'Connection': 'close'} + if 'headers' not in kwargs + else kwargs['headers'] + ) body = b'' if 'body' not in kwargs else kwargs['body'] crlf = '\r\n' @@ -303,13 +340,16 @@ class TestUnitHTTP(TestUnit): sock_types = { 'ipv4': socket.AF_INET, 'ipv6': socket.AF_INET6, - 'unix': socket.AF_UNIX + 'unix': socket.AF_UNIX, } if 'sock' not in kwargs: sock = socket.socket(sock_types[sock_type], socket.SOCK_STREAM) - if sock_type == sock_types['ipv4'] or sock_type == sock_types['ipv6']: + if ( + sock_type == sock_types['ipv4'] + or sock_type == sock_types['ipv6'] + ): sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) if 'wrapper' in kwargs: @@ -357,7 +397,9 @@ class TestUnitHTTP(TestUnit): if 'no_recv' not in kwargs: enc = 'utf-8' if 'encoding' not in kwargs else kwargs['encoding'] - read_timeout = 5 if 'read_timeout' not in kwargs else kwargs['read_timeout'] + read_timeout = ( + 5 if 'read_timeout' not in kwargs else kwargs['read_timeout'] + ) resp = self.recvall(sock, read_timeout=read_timeout).decode(enc) if TestUnit.detailed: @@ -410,7 +452,9 @@ class TestUnitHTTP(TestUnit): p = re.compile('(.*?)\x0d\x0a?', re.M | re.S) headers_lines = p.findall(headers_text) - status = re.search('^HTTP\/\d\.\d\s(\d+)|$', headers_lines.pop(0)).group(1) + status = re.search( + '^HTTP\/\d\.\d\s(\d+)|$', headers_lines.pop(0) + ).group(1) headers = {} for line in headers_lines: @@ -418,16 +462,15 @@ class TestUnitHTTP(TestUnit): if m.group(1) not in headers: headers[m.group(1)] = m.group(2) + elif isinstance(headers[m.group(1)], list): headers[m.group(1)].append(m.group(2)) + else: headers[m.group(1)] = [headers[m.group(1)], m.group(2)] - return { - 'status': int(status), - 'headers': headers, - 'body': body - } + return {'status': int(status), 'headers': headers, 'body': body} + class TestUnitControl(TestUnitHTTP): @@ -441,32 +484,39 @@ class TestUnitControl(TestUnitHTTP): if path[:1] != '/': path = '/config/' + path - return json.loads(self.put( - url=path, - body=conf, - sock_type='unix', - addr=self.testdir + '/control.unit.sock' - )['body']) + return json.loads( + self.put( + url=path, + body=conf, + sock_type='unix', + addr=self.testdir + '/control.unit.sock', + )['body'] + ) def conf_get(self, path='/config'): if path[:1] != '/': path = '/config/' + path - return json.loads(self.get( - url=path, - sock_type='unix', - addr=self.testdir + '/control.unit.sock' - )['body']) + return json.loads( + self.get( + url=path, + sock_type='unix', + addr=self.testdir + '/control.unit.sock', + )['body'] + ) def conf_delete(self, path='/config'): if path[:1] != '/': path = '/config/' + path - return json.loads(self.delete( - url=path, - sock_type='unix', - addr=self.testdir + '/control.unit.sock' - )['body']) + return json.loads( + self.delete( + url=path, + sock_type='unix', + addr=self.testdir + '/control.unit.sock', + )['body'] + ) + class TestUnitApplicationProto(TestUnitControl): @@ -482,64 +532,68 @@ class TestUnitApplicationProto(TestUnitControl): with open(self.testdir + '/unit.log', 'r', errors='ignore') as f: return re.search(pattern, f.read()) + class TestUnitApplicationPython(TestUnitApplicationProto): def load(self, script, name=None): if name is None: name = script - self.conf({ - "listeners": { - "*:7080": { - "application": name - } - }, - "applications": { - name: { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/' + script, - "working_directory": self.current_dir + '/python/' + script, - "module": "wsgi" - } + script_path = self.current_dir + '/python/' + script + + self.conf( + { + "listeners": {"*:7080": {"application": name}}, + "applications": { + name: { + "type": "python", + "processes": {"spare": 0}, + "path": script_path, + "working_directory": script_path, + "module": "wsgi", + } + }, } - }) + ) + class TestUnitApplicationRuby(TestUnitApplicationProto): def load(self, script, name='config.ru'): - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "type": "ruby", - "processes": { "spare": 0 }, - "working_directory": self.current_dir + '/ruby/' + script, - "script": self.current_dir + '/ruby/' + script + '/' + name - } + script_path = self.current_dir + '/ruby/' + script + + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "type": "ruby", + "processes": {"spare": 0}, + "working_directory": script_path, + "script": script_path + '/' + name, + } + }, } - }) + ) + class TestUnitApplicationPHP(TestUnitApplicationProto): def load(self, script, name='index.php'): - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "type": "php", - "processes": { "spare": 0 }, - "root": self.current_dir + '/php/' + script, - "working_directory": self.current_dir + '/php/' + script, - "index": name - } + script_path = self.current_dir + '/php/' + script + + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "type": "php", + "processes": {"spare": 0}, + "root": script_path, + "working_directory": script_path, + "index": name, + } + }, } - }) + ) + class TestUnitApplicationGo(TestUnitApplicationProto): def load(self, script, name='app'): @@ -547,58 +601,67 @@ class TestUnitApplicationGo(TestUnitApplicationProto): if not os.path.isdir(self.testdir + '/go'): os.mkdir(self.testdir + '/go') + go_app_path = self.current_dir + '/go/' + env = os.environ.copy() env['GOPATH'] = self.pardir + '/go' - process = subprocess.Popen(['go', 'build', '-o', - self.testdir + '/go/' + name, - self.current_dir + '/go/' + script + '/' + name + '.go'], - env=env) + process = subprocess.Popen( + [ + 'go', + 'build', + '-o', + self.testdir + '/go/' + name, + go_app_path + script + '/' + name + '.go', + ], + env=env, + ) process.communicate() - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "type": "external", - "processes": { "spare": 0 }, - "working_directory": self.current_dir + '/go/' + script, - "executable": self.testdir + '/go/' + name - } + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "type": "external", + "processes": {"spare": 0}, + "working_directory": go_app_path + script, + "executable": self.testdir + '/go/' + name, + } + }, } - }) + ) + class TestUnitApplicationNode(TestUnitApplicationProto): def load(self, script, name='app.js'): # copy application - shutil.copytree(self.current_dir + '/node/' + script, - self.testdir + '/node') + shutil.copytree( + self.current_dir + '/node/' + script, self.testdir + '/node' + ) # link modules - os.symlink(self.pardir + '/node/node_modules', - self.testdir + '/node/node_modules') - - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "type": "external", - "processes": { "spare": 0 }, - "working_directory": self.testdir + '/node', - "executable": name - } + os.symlink( + self.pardir + '/node/node_modules', + self.testdir + '/node/node_modules', + ) + + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "type": "external", + "processes": {"spare": 0}, + "working_directory": self.testdir + '/node', + "executable": name, + } + }, } - }) + ) + class TestUnitApplicationJava(TestUnitApplicationProto): def load(self, script, name='app'): @@ -641,48 +704,53 @@ class TestUnitApplicationJava(TestUnitApplicationProto): if not os.path.isdir(classes_path): os.makedirs(classes_path) - javac = ['javac', '-encoding', 'utf-8', '-d', classes_path, - '-classpath', - self.pardir + '/build/tomcat-servlet-api-9.0.13.jar'] + tomcat_jar = self.pardir + '/build/tomcat-servlet-api-9.0.13.jar' + + javac = [ + 'javac', + '-encoding', 'utf-8', + '-d', classes_path, + '-classpath', tomcat_jar, + ] javac.extend(src) process = subprocess.Popen(javac) process.communicate() - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "unit_jars": self.pardir + '/build', - "type": "java", - "processes": { "spare": 0 }, - "working_directory": script_path, - "webapp": app_path - } + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "unit_jars": self.pardir + '/build', + "type": "java", + "processes": {"spare": 0}, + "working_directory": script_path, + "webapp": app_path, + } + }, } - }) + ) + class TestUnitApplicationPerl(TestUnitApplicationProto): def load(self, script, name='psgi.pl'): - self.conf({ - "listeners": { - "*:7080": { - "application": script - } - }, - "applications": { - script: { - "type": "perl", - "processes": { "spare": 0 }, - "working_directory": self.current_dir + '/perl/' + script, - "script": self.current_dir + '/perl/' + script + '/' + name - } + script_path = self.current_dir + '/perl/' + script + + self.conf( + { + "listeners": {"*:7080": {"application": script}}, + "applications": { + script: { + "type": "perl", + "processes": {"spare": 0}, + "working_directory": script_path, + "script": script_path + '/' + name, + } + }, } - }) + ) + class TestUnitApplicationTLS(TestUnitApplicationProto): def __init__(self, test): @@ -693,10 +761,18 @@ class TestUnitApplicationTLS(TestUnitApplicationProto): self.context.verify_mode = ssl.CERT_NONE def certificate(self, name='default', load=True): - subprocess.call(['openssl', 'req', '-x509', '-new', '-config', - self.testdir + '/openssl.conf', '-subj', '/CN=' + name + '/', - '-out', self.testdir + '/' + name + '.crt', - '-keyout', self.testdir + '/' + name + '.key']) + subprocess.call( + [ + 'openssl', + 'req', + '-x509', + '-new', + '-subj', '/CN=' + name + '/', + '-config', self.testdir + '/openssl.conf', + '-out', self.testdir + '/' + name + '.crt', + '-keyout', self.testdir + '/' + name + '.key', + ] + ) if load: self.certificate_load(name) @@ -705,17 +781,17 @@ class TestUnitApplicationTLS(TestUnitApplicationProto): if key is None: key = crt - with open(self.testdir + '/' + key + '.key', 'rb') as k, \ - open(self.testdir + '/' + crt + '.crt', 'rb') as c: - return self.conf(k.read() + c.read(), '/certificates/' + crt) + key_path = self.testdir + '/' + key + '.key' + crt_path = self.testdir + '/' + crt + '.crt' + + with open(key_path, 'rb') as k, open(crt_path, 'rb') as c: + return self.conf(k.read() + c.read(), '/certificates/' + crt) def get_ssl(self, **kwargs): - return self.get(wrapper=self.context.wrap_socket, - **kwargs) + return self.get(wrapper=self.context.wrap_socket, **kwargs) def post_ssl(self, **kwargs): - return self.post(wrapper=self.context.wrap_socket, - **kwargs) + return self.post(wrapper=self.context.wrap_socket, **kwargs) def get_server_certificate(self, addr=('127.0.0.1', 7080)): @@ -739,25 +815,27 @@ class TestUnitApplicationTLS(TestUnitApplicationProto): # create default openssl configuration with open(self.testdir + '/openssl.conf', 'w') as f: - f.write("""[ req ] + f.write( + """[ req ] default_bits = 1024 encrypt_key = no distinguished_name = req_distinguished_name -[ req_distinguished_name ]""") - - self.conf({ - "listeners": { - "*:7080": { - "application": name - } - }, - "applications": { - name: { - "type": "python", - "processes": { "spare": 0 }, - "path": self.current_dir + '/python/' + script, - "working_directory": self.current_dir + '/python/' + script, - "module": "wsgi" - } +[ req_distinguished_name ]""" + ) + + script_path = self.current_dir + '/python/' + script + + self.conf( + { + "listeners": {"*:7080": {"application": name}}, + "applications": { + name: { + "type": "python", + "processes": {"spare": 0}, + "path": script_path, + "working_directory": script_path, + "module": "wsgi", + } + }, } - }) + ) |