summaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
authorAndrey Zelenkov <zelenkov@nginx.com>2019-03-26 23:38:30 +0300
committerAndrey Zelenkov <zelenkov@nginx.com>2019-03-26 23:38:30 +0300
commit281899fcef10eaf815d90958d49243c5060ffac0 (patch)
tree258d01fb832a4253815db41673c9fcc73f128e6b
parent3d7a47c9acb1120f90225c833fd56cffeb99c2cd (diff)
downloadunit-281899fcef10eaf815d90958d49243c5060ffac0.tar.gz
unit-281899fcef10eaf815d90958d49243c5060ffac0.tar.bz2
Tests: style.
Diffstat (limited to '')
-rw-r--r--test/test_access_log.py151
-rw-r--r--test/test_configuration.py400
-rw-r--r--test/test_go_application.py144
-rw-r--r--test/test_http_header.py476
-rw-r--r--test/test_java_application.py1178
-rw-r--r--test/test_node_application.py340
-rw-r--r--test/test_perl_application.py157
-rw-r--r--test/test_php_application.py330
-rw-r--r--test/test_php_basic.py141
-rw-r--r--test/test_python_application.py380
-rw-r--r--test/test_python_basic.py148
-rw-r--r--test/test_python_environment.py215
-rw-r--r--test/test_python_procman.py189
-rw-r--r--test/test_routing.py990
-rw-r--r--test/test_ruby_application.py204
-rw-r--r--test/test_settings.py143
-rw-r--r--test/test_tls.py469
-rw-r--r--test/unit.py484
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",
+ }
+ },
}
- })
+ )