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