summaryrefslogtreecommitdiffhomepage
path: root/test/test_node_application.py
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--test/test_node_application.py340
1 files changed, 221 insertions, 119 deletions
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()