technocloud-public/nginx-unit
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
Files
e2b53e16c60ba1e3bbbe59172c184e97f889326b
nginx-unit/test/test_java_application.py
Tiago Natel de Moura e9e5ddd5a5 Refactor of process management. 
The process abstraction has changed to:

  setup(task, process)
  start(task, process_data)
  prefork(task, process, mp)

The prefork() occurs in the main process right before fork.

The file src/nxt_main_process.c is completely free of process
specific logic.

The creation of a process now supports a PROCESS_CREATED state.  The
The setup() function of each process can set its state to either
created or ready.  If created, a MSG_PROCESS_CREATED is sent to main
process, where external setup can be done (required for rootfs under
container).

The core processes (discovery, controller and router) doesn't need
external setup, then they all proceeds to their start() function
straight away.

In the case of applications, the load of the module happens at the
process setup() time and The module's init() function has changed
to be the start() of the process.

The module API has changed to:

  setup(task, process, conf)
  start(task, data)

As a direct benefit of the PROCESS_CREATED message, the clone(2) of
processes using pid namespaces now doesn't need to create a pipe
to make the child block until parent setup uid/gid mappings nor it
needs to receive the child pid.
2020-03-09 16:28:25 +00:00

1268 lines
38 KiB
Python
Raw Blame History

import io
import os
import time
from unit.applications.lang.java import TestApplicationJava
class TestJavaApplication(TestApplicationJava):
prerequisites = {'modules': {'java': 'all'}}
def test_java_conf_error(self):
self.skip_alerts.extend(
[
r'realpath.*failed',
r'failed to apply new conf',
r'application setup failed',
]
)
self.assertIn(
'error',
self.conf(
{
"listeners": {"*:7080": {"pass": "applications/app"}},
"applications": {
"app": {
"type": "java",
"processes": 1,
"working_directory": self.current_dir
+ "/java/empty",
"webapp": self.testdir + "/java",
"unit_jars": self.testdir + "/no_such_dir",
}
},
}
),
'conf error',
)
def test_java_war(self):
self.load('empty_war')
self.assertIn(
'success',
self.conf(
'"' + self.testdir + '/java/empty.war"',
'/config/applications/empty_war/webapp',
),
'configure war',
)
self.assertEqual(self.get()['status'], 200, 'war')
def test_java_application_cookies(self):
self.load('cookies')
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')
def test_java_application_filter(self):
self.load('filter')
headers = self.get()['headers']
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',
)
def test_java_application_get_variables(self):
self.load('get_params')
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',
)
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']
self.assertEqual(headers['X-Var-1'], 'val1', 'POST variables')
self.assertEqual(headers['X-Var-2'], 'true', 'POST variables 2')
self.assertEqual(headers['X-Var-3'], 'false', 'POST variables 3')
def test_java_application_session(self):
self.load('session')
headers = self.get(url='/?var1=val1')['headers']
session_id = headers['X-Session-Id']
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']
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'
)
def test_java_application_session_active(self):
self.load('session_inactive')
resp = self.get(headers={
'X-Interval': '4',
'Host': 'localhost',
'Connection': 'close',
})
session_id = resp['headers']['X-Session-Id']
self.assertEqual(resp['status'], 200, 'session init')
self.assertEqual(
resp['headers']['X-Session-Interval'], '4', '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',
}
)
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',
}
)
self.assertEqual(
resp['headers']['X-Session-Id'], session_id, 'session active 2'
)
time.sleep(2)
resp = self.get(
headers={
'Host': 'localhost',
'Cookie': 'JSESSIONID=' + session_id,
'Connection': 'close',
}
)
self.assertEqual(
resp['headers']['X-Session-Id'], session_id, 'session active 3'
)
def test_java_application_session_inactive(self):
self.load('session_inactive')
resp = self.get(headers={
'X-Interval': '1',
'Host': 'localhost',
'Connection': 'close',
})
session_id = resp['headers']['X-Session-Id']
time.sleep(3)
resp = self.get(
headers={
'Host': 'localhost',
'Cookie': 'JSESSIONID=' + session_id,
'Connection': 'close',
}
)
self.assertNotEqual(
resp['headers']['X-Session-Id'], session_id, 'session inactive'
)
def test_java_application_session_invalidate(self):
self.load('session_invalidate')
resp = self.get()
session_id = resp['headers']['X-Session-Id']
resp = self.get(
headers={
'Host': 'localhost',
'Cookie': 'JSESSIONID=' + session_id,
'Connection': 'close',
}
)
self.assertNotEqual(
resp['headers']['X-Session-Id'], session_id, 'session invalidate'
)
def test_java_application_session_listeners(self):
self.load('session_listeners')
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'
)
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'
)
def test_java_application_jsp(self):
self.load('jsp')
headers = self.get(url='/index.jsp')['headers']
self.assertEqual(headers['X-Unit-JSP'], 'ok', 'JSP Ok header')
def test_java_application_url_pattern(self):
self.load('url_pattern')
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-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-Path-Info'], '/index.bop', '#2 path info')
headers = self.get(url='/baz')['headers']
self.assertEqual(headers['X-Id'], 'servlet2', '#3 Servlet2 request')
self.assertEqual(headers['X-Request-URI'], '/baz', '#3 request URI')
self.assertEqual(headers['X-Servlet-Path'], '/baz', '#3 servlet path')
self.assertEqual(headers['X-Path-Info'], 'null', '#3 path info')
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-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-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-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-Path-Info'], 'null', '#7 path info')
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-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-Servlet-Path'], '/foo', '#9 servlet path')
self.assertEqual(headers['X-Path-Info'], '/baz', '#9 path info')
headers = self.get()['headers']
self.assertEqual(headers['X-Id'], 'default', '#10 default request')
self.assertEqual(headers['X-Request-URI'], '/', '#10 request URI')
self.assertEqual(headers['X-Servlet-Path'], '/', '#10 servlet path')
self.assertEqual(headers['X-Path-Info'], 'null', '#10 path info')
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-Path-Info'], 'null', '#11 path info')
def test_java_application_header(self):
self.load('header')
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-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')
def test_java_application_content_type(self):
self.load('content_type')
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'
)
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',
)
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',
)
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',
)
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',
)
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'
)
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'
)
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'
)
def test_java_application_welcome_files(self):
self.load('welcome_files')
headers = self.get()['headers']
resp = self.get(url='/dir1')
self.assertEqual(resp['status'], 302, 'dir redirect expected')
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'
)
headers = self.get(url='/dir2/')['headers']
self.assertEqual(headers['X-Unit-JSP'], 'ok', 'JSP Ok header')
self.assertEqual(headers['X-JSP-Filter'], '1', 'JSP Filter header')
headers = self.get(url='/dir3/')['headers']
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',
)
headers = self.get(url='/dir5/')['headers']
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'
)
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'
)
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'
)
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',
)
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'
)
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',
)
def test_java_application_named_dispatcher_forward(self):
self.load('forward')
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',
)
def test_java_application_request_uri_include(self):
self.load('include')
resp = self.get(url='/inc?uri=/data/test')
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'
)
def test_java_application_named_dispatcher_include(self):
self.load('include')
resp = self.get(url='/inc?disp=name&uri=data')
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'
)
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'],
headers['X-Real-Path'] + headers['X-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',
)
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'
)
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',
)
def test_java_application_query_empty(self):
self.load('query_string')
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',
)
def test_java_application_empty(self):
self.load('empty')
self.assertEqual(self.get()['status'], 200, 'empty')
def test_java_application_keepalive_body(self):
self.load('mirror')
self.assertEqual(self.post()['status'], 200, 'init')
body = '0123456789' * 500
(resp, sock) = self.post(
headers={
'Connection': 'keep-alive',
'Content-Type': 'text/html',
'Host': 'localhost',
},
start=True,
body=body,
read_timeout=1,
)
self.assertEqual(resp['body'], body, 'keep-alive 1')
body = '0123456789'
resp = self.post(
headers={
'Connection': 'close',
'Content-Type': 'text/html',
'Host': 'localhost',
},
sock=sock,
body=body,
)
self.assertEqual(resp['body'], body, 'keep-alive 2')
def test_java_application_http_10(self):
self.load('empty')
self.assertEqual(self.get(http_10=True)['status'], 200, 'HTTP 1.0')
def test_java_application_no_method(self):
self.load('empty')
self.assertEqual(self.post()['status'], 405, 'no method')
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',
'Connection': 'close',
}
)['headers']['X-Reply'],
'blah',
'get header',
)
def test_java_application_get_header_empty(self):
self.load('get_header')
self.assertNotIn('X-Reply', self.get()['headers'], 'get header empty')
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',
'Connection': 'close',
}
)['headers']
self.assertEqual(headers['X-Reply-0'], 'blah', 'get headers')
self.assertEqual(headers['X-Reply-1'], 'blah', 'get headers 2')
def test_java_application_get_headers_empty(self):
self.load('get_headers')
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',
)
def test_java_application_header_int(self):
self.load('header_int')
headers = self.get(
headers={
'X-Header': '2',
'Content-Type': 'text/html',
'Host': 'localhost',
'Connection': 'close',
}
)['headers']
self.assertEqual(headers['X-Set-Int'], '1', 'set int header')
self.assertEqual(headers['X-Get-Int'], '2', 'get int header')
def test_java_application_header_date(self):
self.load('header_date')
date = 'Fri, 15 Mar 2019 14:45:34 GMT'
headers = self.get(
headers={
'X-Header': date,
'Content-Type': 'text/html',
'Host': 'localhost',
'Connection': 'close',
}
)['headers']
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')
def test_java_application_multipart(self):
self.load('multipart')
reldst = '/uploads'
fulldst = self.testdir + reldst
os.mkdir(fulldst)
self.public_dir(fulldst)
fields = {
'file': {
'filename': 'sample.txt',
'type': 'text/plain',
'data': io.StringIO('Data from sample file'),
},
'destination': fulldst,
'upload': 'Upload',
}
encoded, content_type = self.multipart_encode(fields)
preamble = 'Preamble. Should be ignored.'
epilogue = 'Epilogue. Should be ignored.'
body = "%s\r\n%s\r\n%s" % (preamble, encoded.decode(), epilogue)
resp = self.post(
headers={
'Content-Type': content_type,
'Host': 'localhost',
'Connection': 'close',
},
body=body,
)
self.assertEqual(resp['status'], 200, 'multipart status')
self.assertRegex(
resp['body'], r'sample\.txt created', 'multipart body'
)
self.assertIsNotNone(
self.search_in_log(
r'^Data from sample file$', name=reldst + '/sample.txt'
),
'file created',
)
if __name__ == '__main__':
TestJavaApplication.main()
Reference in New Issue View Git Blame Copy Permalink