Skip to content
Snippets Groups Projects
Commit d8c09d07 authored by slobinger's avatar slobinger
Browse files

Merge branch '9-sort-test_sams_proxy' into 'master' 

removed eprints and sorted test methods in test_sams_proxy.py fixed #9

Closes #9

See merge request !4
parents 84e849ed df9c62d1
Branches
Tags
1 merge request!4removed eprints and sorted test methods in test_sams_proxy.py fixed #9
Hide whitespace changes
Inline Side-by-side
test/api_test_server.py
+ 0
− 1
View file @ d8c09d07
...@@ -16,7 +16,6 @@ def hello(): ...@@ -16,7 +16,6 @@ def hello():
@app.route('/passthrough', @app.route('/passthrough',
methods=['GET', 'POST', 'DELETE', 'UPDATE', 'PATCH', 'OPTIONS', 'PUT', 'HEAD']) methods=['GET', 'POST', 'DELETE', 'UPDATE', 'PATCH', 'OPTIONS', 'PUT', 'HEAD'])
def passthrough(): def passthrough():
eprint('request.headers: ' + str(request.headers) + '\n')
body = request.form body = request.form
if request.is_json: if request.is_json:
body = request.get_json() body = request.get_json()
......
test/test_sams_proxy.py
+ 79
− 85
View file @ d8c09d07
...@@ -11,31 +11,19 @@ def eprint(*args, **kwargs): ...@@ -11,31 +11,19 @@ def eprint(*args, **kwargs):
print(*args, file=sys.stderr, **kwargs) print(*args, file=sys.stderr, **kwargs)
class TestSAMSHub(unittest.TestCase): class TestSAMSHub(unittest.TestCase):
def setUp(self):
from .api_test_server import app as apiApp
#apiApp.config['TESTING'] = True
thApi = FlaskInThread(apiApp, host="localhost", port=4711)
thApi.start()
time.sleep(0.01)
def tearDown(self):
response = requests.get('http://localhost:4711/shutdown')
print(response.content.decode('utf-8'))
time.sleep(0.01)
def test_proxy_spec_mandatory (self): def test_proxy_spec_mandatory (self):
""" SAMSProxy needs a proxy_spec dict, raies error if not supplied""" """ SAMSProxy needs a proxy_spec dict, raies error if not supplied"""
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
SAMSProxy() SAMSProxy()
with self.assertRaises(ProxySpecMissing): with self.assertRaises(ProxySpecMissing):
SAMSProxy(proxySpec = {}) SAMSProxy(proxySpec = {})
def test_proxy_spec_musst_be_dict(self): def test_proxy_spec_musst_be_dict(self):
""" SAMSProxy raies an TypeExecption if ProxySpec is no dict""" """ SAMSProxy raies an TypeExecption if ProxySpec is no dict"""
with self.assertRaises(TypeError): with self.assertRaises(TypeError):
SAMSProxy(proxySpec = 'wrong') SAMSProxy(proxySpec = 'wrong')
def test_mandatory_keys_in_spec(self): def test_mandatory_keys_in_spec(self):
""" ProxySpecKeyMissing is raised if something is mising in spec""" """ ProxySpecKeyMissing is raised if something is mising in spec"""
with self.assertRaises(ProxySpecKeyMssing): with self.assertRaises(ProxySpecKeyMssing):
...@@ -62,80 +50,79 @@ class TestSAMSHub(unittest.TestCase): ...@@ -62,80 +50,79 @@ class TestSAMSHub(unittest.TestCase):
, 'session_passthrough': {'default': { , 'session_passthrough': {'default': {
'user': {'param_type': 'body', 'name': 'u'} 'user': {'param_type': 'body', 'name': 'u'}
}, method: {'user': {'name': 'u'}}}}) }, method: {'user': {'name': 'u'}}}})
def test_simple_proxy_forwarding(self): class TestSAMSHubWithThreadedAPI(unittest.TestCase):
""" The proxy forwards a request to test flaskapp and returns 'hello' """
proxyApp = Flask('test') def setUp(self):
proxyApp.config['TESTING'] = True from .api_test_server import app as apiApp
proxy = SAMSProxy( apiApp.config['TESTING'] = True
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'}) thApi = FlaskInThread(apiApp, host="localhost", port=4711)
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', thApi.start()
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', time.sleep(0.01)
'PATCH', 'DELETE', 'OPTIONS']) self.proxyApp = Flask('test')
with proxyApp.test_client(self) as client: self.proxyApp.config['TESTING'] = True
for method in ['get', 'post', 'put', 'update', 'patch', 'delete',
'options']: def tearDown(self):
response = client.open(path='/proxy/hello', method=method.upper()) response = requests.get('http://localhost:4711/shutdown')
self.assertIs(response.status_code, 200, method) print(response.content.decode('utf-8'))
self.assertEqual(response.data.decode('utf-8'), method + ' hello') time.sleep(0.01)
def test_param_passthrough(self): def test_json_passthrough(self):
"""The proxy passes headers, form / data and url params """ """The proxy passes json data """
proxyApp = Flask('test')
proxyApp.config['TESTING'] = True
proxy = SAMSProxy( proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'}) proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'})
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS']) 'PATCH', 'DELETE', 'OPTIONS'])
with proxyApp.test_client(self) as client: thProxyApp = FlaskInThread(self.proxyApp, host="localhost", port=5000)
res = client.get(path='/proxy/passthrough', thProxyApp.start()
query_string={'u':'foo'}, data={'g':'bar'}, headers={'lang': 'klingon'}) time.sleep(0.01)
resDict = json.loads(res.data.decode('utf-8')) res = requests.post(url='http://localhost:5000/proxy/passthrough',
self.assertEqual(resDict['url']['u'], 'foo') json={'g':'bar'})
self.assertEqual(resDict['body']['g'], 'bar') resDict = json.loads(res.content.decode('utf-8'))
self.assertEqual(resDict['header']['Lang'], 'klingon') self.assertEqual(resDict['body']['g'], 'bar')
requests.get('http://localhost:5000/shutdown')
def test_session_passthrough(self): time.sleep(0.01)
"""The proxy passes the specified session param as specified param_type"""
proxyApp = Flask('test') def test_session_wins(self):
proxyApp.config['TESTING'] = True """proxy guaranties that session params will overwrite the request params"""
proxy = SAMSProxy( proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711', proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': { 'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'}, 'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'}, 'group': {'name': 'g', 'param-type': 'body'},
'language': {'name': 'lang', 'param-type': 'header'}}}}) 'language': {'name': 'Lang', 'param-type': 'header'}}}})
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS']) 'PATCH', 'DELETE', 'OPTIONS'])
proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT' self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
with proxyApp.test_client(self) as client: with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess: with client.session_transaction() as sess:
sess['user'] = 'foo' sess['user'] = 'foo'
sess['group'] = 'bar' sess['group'] = 'bar'
sess['language'] = 'klingon' sess['language'] = 'klingon'
res = client.get(path='/proxy/passthrough') res = client.get(path='/proxy/passthrough', query_string={'u':'quin'},
data={'g':'admins'}, headers={'Lang': 'romulan'})
resDict = json.loads(res.data.decode('utf-8')) resDict = json.loads(res.data.decode('utf-8'))
self.assertEqual(resDict['url']['u'], sess['user']) self.assertEqual(resDict['url']['u'], sess['user'])
self.assertEqual(resDict['body']['g'], sess['group']) self.assertEqual(resDict['body']['g'], sess['group'])
self.assertEqual(resDict['header']['Lang'], sess['language']) self.assertEqual(resDict['header']['Lang'], sess['language'])
def test_session_param_passthrough(self): def test_session_param_passthrough(self):
"""The proxy passes the session and the normal params""" """The proxy passes the session and the normal params"""
proxyApp = Flask('test') self.proxyApp = Flask('test')
proxyApp.config['TESTING'] = True self.proxyApp.config['TESTING'] = True
proxy = SAMSProxy( proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711', proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': { 'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'}, 'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'}, 'group': {'name': 'g', 'param-type': 'body'},
'language': {'name': 'lang', 'param-type': 'header'}}}}) 'language': {'name': 'lang', 'param-type': 'header'}}}})
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS']) 'PATCH', 'DELETE', 'OPTIONS'])
proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT' self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
with proxyApp.test_client(self) as client: with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess: with client.session_transaction() as sess:
sess['user'] = 'foo' sess['user'] = 'foo'
sess['group'] = 'bar' sess['group'] = 'bar'
...@@ -149,48 +136,55 @@ class TestSAMSHub(unittest.TestCase): ...@@ -149,48 +136,55 @@ class TestSAMSHub(unittest.TestCase):
self.assertEqual(resDict['url']['q'], 'Query') self.assertEqual(resDict['url']['q'], 'Query')
self.assertEqual(resDict['body']['f'], 'Form') self.assertEqual(resDict['body']['f'], 'Form')
self.assertEqual(resDict['header']['Some-Mode'], 'whatever') self.assertEqual(resDict['header']['Some-Mode'], 'whatever')
def test_session_wins(self): def test_session_passthrough(self):
"""proxy guaranties that session params will overwrite the request params""" """The proxy passes the specified session param as specified param_type"""
proxyApp = Flask('test')
proxyApp.config['TESTING'] = True
proxy = SAMSProxy( proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711', proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': { 'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'}, 'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'}, 'group': {'name': 'g', 'param-type': 'body'},
'language': {'name': 'Lang', 'param-type': 'header'}}}}) 'language': {'name': 'lang', 'param-type': 'header'}}}})
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS']) 'PATCH', 'DELETE', 'OPTIONS'])
proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT' self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
with proxyApp.test_client(self) as client: with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess: with client.session_transaction() as sess:
sess['user'] = 'foo' sess['user'] = 'foo'
sess['group'] = 'bar' sess['group'] = 'bar'
sess['language'] = 'klingon' sess['language'] = 'klingon'
res = client.get(path='/proxy/passthrough', query_string={'u':'quin'}, res = client.get(path='/proxy/passthrough')
data={'g':'admins'}, headers={'Lang': 'romulan'})
resDict = json.loads(res.data.decode('utf-8')) resDict = json.loads(res.data.decode('utf-8'))
self.assertEqual(resDict['url']['u'], sess['user']) self.assertEqual(resDict['url']['u'], sess['user'])
self.assertEqual(resDict['body']['g'], sess['group']) self.assertEqual(resDict['body']['g'], sess['group'])
self.assertEqual(resDict['header']['Lang'], sess['language']) self.assertEqual(resDict['header']['Lang'], sess['language'])
def test_json_passthrough(self): def test_simple_proxy_forwarding(self):
"""The proxy passes json data """ """ The proxy forwards a request to test flaskapp and returns 'hello' """
proxyApp = Flask('test')
proxy = SAMSProxy( proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'}) proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'})
proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy', self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE', view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS']) 'PATCH', 'DELETE', 'OPTIONS'])
thProxyApp = FlaskInThread(proxyApp, host="localhost", port=5000) with self.proxyApp.test_client(self) as client:
thProxyApp.start() for method in ['get', 'post', 'put', 'update', 'patch', 'delete',
time.sleep(1) 'options']:
res = requests.post(url='http://localhost:5000/proxy/passthrough', response = client.open(path='/proxy/hello', method=method.upper())
json={'g':'bar'}) self.assertIs(response.status_code, 200, method)
resDict = json.loads(res.content.decode('utf-8')) self.assertEqual(response.data.decode('utf-8'), method + ' hello')
eprint('resDict: ' + str(resDict) + '\n')
self.assertEqual(resDict['body']['g'], 'bar') def test_param_passthrough(self):
requests.get('http://localhost:5000/shutdown') """The proxy passes headers, form / data and url params """
time.sleep(1) proxy = SAMSProxy(
\ No newline at end of file proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'})
self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS'])
with self.proxyApp.test_client(self) as client:
res = client.get(path='/proxy/passthrough',
query_string={'u':'foo'}, data={'g':'bar'}, headers={'lang': 'klingon'})
resDict = json.loads(res.data.decode('utf-8'))
self.assertEqual(resDict['url']['u'], 'foo')
self.assertEqual(resDict['body']['g'], 'bar')
self.assertEqual(resDict['header']['Lang'], 'klingon')
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment