diff --git a/test/api_test_server.py b/test/api_test_server.py
index c8cd6eeb515bc27bf5eba91ee43f3a3de8f22944..ed38380313c21eeb8e9995c8aa8c31ad56e0cf9f 100644
--- a/test/api_test_server.py
+++ b/test/api_test_server.py
@@ -16,7 +16,6 @@ def hello():
@app.route('/passthrough',
methods=['GET', 'POST', 'DELETE', 'UPDATE', 'PATCH', 'OPTIONS', 'PUT', 'HEAD'])
def passthrough():
- eprint('request.headers: ' + str(request.headers) + '\n')
body = request.form
if request.is_json:
body = request.get_json()
diff --git a/test/test_sams_proxy.py b/test/test_sams_proxy.py
index 1a4c8fe804c8baa50a7875091d9a82c2d9183c33..ed1099a2e59231b0865d7c413784cc5eb1ec175f 100644
--- a/test/test_sams_proxy.py
+++ b/test/test_sams_proxy.py
@@ -11,31 +11,19 @@ def eprint(*args, **kwargs):
print(*args, file=sys.stderr, **kwargs)
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):
""" SAMSProxy needs a proxy_spec dict, raies error if not supplied"""
with self.assertRaises(TypeError):
SAMSProxy()
with self.assertRaises(ProxySpecMissing):
SAMSProxy(proxySpec = {})
-
+
def test_proxy_spec_musst_be_dict(self):
""" SAMSProxy raies an TypeExecption if ProxySpec is no dict"""
with self.assertRaises(TypeError):
SAMSProxy(proxySpec = 'wrong')
-
+
def test_mandatory_keys_in_spec(self):
""" ProxySpecKeyMissing is raised if something is mising in spec"""
with self.assertRaises(ProxySpecKeyMssing):
@@ -62,80 +50,79 @@ class TestSAMSHub(unittest.TestCase):
, 'session_passthrough': {'default': {
'user': {'param_type': 'body', 'name': 'u'}
}, method: {'user': {'name': 'u'}}}})
-
- def test_simple_proxy_forwarding(self):
- """ The proxy forwards a request to test flaskapp and returns 'hello' """
- proxyApp = Flask('test')
- proxyApp.config['TESTING'] = True
- proxy = SAMSProxy(
- proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711'})
- proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
- view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
- 'PATCH', 'DELETE', 'OPTIONS'])
- with proxyApp.test_client(self) as client:
- for method in ['get', 'post', 'put', 'update', 'patch', 'delete',
- 'options']:
- response = client.open(path='/proxy/hello', method=method.upper())
- self.assertIs(response.status_code, 200, method)
- self.assertEqual(response.data.decode('utf-8'), method + ' hello')
-
- def test_param_passthrough(self):
- """The proxy passes headers, form / data and url params """
- proxyApp = Flask('test')
- proxyApp.config['TESTING'] = True
+
+class TestSAMSHubWithThreadedAPI(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)
+ self.proxyApp = Flask('test')
+ self.proxyApp.config['TESTING'] = True
+
+ def tearDown(self):
+ response = requests.get('http://localhost:4711/shutdown')
+ print(response.content.decode('utf-8'))
+ time.sleep(0.01)
+
+ def test_json_passthrough(self):
+ """The proxy passes json data """
proxy = SAMSProxy(
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',
'PATCH', 'DELETE', 'OPTIONS'])
- with 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')
-
- def test_session_passthrough(self):
- """The proxy passes the specified session param as specified param_type"""
- proxyApp = Flask('test')
- proxyApp.config['TESTING'] = True
+ thProxyApp = FlaskInThread(self.proxyApp, host="localhost", port=5000)
+ thProxyApp.start()
+ time.sleep(0.01)
+ res = requests.post(url='http://localhost:5000/proxy/passthrough',
+ json={'g':'bar'})
+ resDict = json.loads(res.content.decode('utf-8'))
+ self.assertEqual(resDict['body']['g'], 'bar')
+ requests.get('http://localhost:5000/shutdown')
+ time.sleep(0.01)
+
+ def test_session_wins(self):
+ """proxy guaranties that session params will overwrite the request params"""
proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'},
- 'language': {'name': 'lang', 'param-type': 'header'}}}})
- proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
+ 'language': {'name': 'Lang', 'param-type': 'header'}}}})
+ self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS'])
- proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
- with proxyApp.test_client(self) as client:
+ self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
+ with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess:
sess['user'] = 'foo'
sess['group'] = 'bar'
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'))
self.assertEqual(resDict['url']['u'], sess['user'])
self.assertEqual(resDict['body']['g'], sess['group'])
self.assertEqual(resDict['header']['Lang'], sess['language'])
-
+
def test_session_param_passthrough(self):
"""The proxy passes the session and the normal params"""
- proxyApp = Flask('test')
- proxyApp.config['TESTING'] = True
+ self.proxyApp = Flask('test')
+ self.proxyApp.config['TESTING'] = True
proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'},
'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',
'PATCH', 'DELETE', 'OPTIONS'])
- proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
- with proxyApp.test_client(self) as client:
+ self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
+ with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess:
sess['user'] = 'foo'
sess['group'] = 'bar'
@@ -149,48 +136,55 @@ class TestSAMSHub(unittest.TestCase):
self.assertEqual(resDict['url']['q'], 'Query')
self.assertEqual(resDict['body']['f'], 'Form')
self.assertEqual(resDict['header']['Some-Mode'], 'whatever')
-
- def test_session_wins(self):
- """proxy guaranties that session params will overwrite the request params"""
- proxyApp = Flask('test')
- proxyApp.config['TESTING'] = True
+
+ def test_session_passthrough(self):
+ """The proxy passes the specified session param as specified param_type"""
proxy = SAMSProxy(
proxySpec = {'in': '/proxy', 'out': 'http://localhost:4711',
'session_passthrough': {'default': {
'user': {'name': 'u','param-type': 'url'},
'group': {'name': 'g', 'param-type': 'body'},
- 'language': {'name': 'Lang', 'param-type': 'header'}}}})
- proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
+ 'language': {'name': 'lang', 'param-type': 'header'}}}})
+ self.proxyApp.add_url_rule(rule = proxy.urlRule, endpoint = 'proxy',
view_func = proxy.proxy, methods=['GET', 'POST', 'PUT', 'UPDATE',
'PATCH', 'DELETE', 'OPTIONS'])
- proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
- with proxyApp.test_client(self) as client:
+ self.proxyApp.secret_key = 'A0Zr98j/3yX R~XHH!jmN]LWX/,?RT'
+ with self.proxyApp.test_client(self) as client:
with client.session_transaction() as sess:
sess['user'] = 'foo'
sess['group'] = 'bar'
sess['language'] = 'klingon'
- res = client.get(path='/proxy/passthrough', query_string={'u':'quin'},
- data={'g':'admins'}, headers={'Lang': 'romulan'})
+ res = client.get(path='/proxy/passthrough')
resDict = json.loads(res.data.decode('utf-8'))
self.assertEqual(resDict['url']['u'], sess['user'])
self.assertEqual(resDict['body']['g'], sess['group'])
self.assertEqual(resDict['header']['Lang'], sess['language'])
- def test_json_passthrough(self):
- """The proxy passes json data """
- proxyApp = Flask('test')
+ def test_simple_proxy_forwarding(self):
+ """ The proxy forwards a request to test flaskapp and returns 'hello' """
proxy = SAMSProxy(
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',
'PATCH', 'DELETE', 'OPTIONS'])
- thProxyApp = FlaskInThread(proxyApp, host="localhost", port=5000)
- thProxyApp.start()
- time.sleep(1)
- res = requests.post(url='http://localhost:5000/proxy/passthrough',
- json={'g':'bar'})
- resDict = json.loads(res.content.decode('utf-8'))
- eprint('resDict: ' + str(resDict) + '\n')
- self.assertEqual(resDict['body']['g'], 'bar')
- requests.get('http://localhost:5000/shutdown')
- time.sleep(1)
\ No newline at end of file
+ with self.proxyApp.test_client(self) as client:
+ for method in ['get', 'post', 'put', 'update', 'patch', 'delete',
+ 'options']:
+ response = client.open(path='/proxy/hello', method=method.upper())
+ self.assertIs(response.status_code, 200, method)
+ self.assertEqual(response.data.decode('utf-8'), method + ' hello')
+
+ def test_param_passthrough(self):
+ """The proxy passes headers, form / data and url params """
+ proxy = SAMSProxy(
+ 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