<s> """<STR_LIT>""" <EOL> import json <EOL> import os <EOL> from queue import Queue <EOL> from subprocess import Popen <EOL> from urllib . parse import urlparse <EOL> from . . import orm <EOL> from . mocking import MockHub <EOL> from . test_api import api_request <EOL> from . . utils import wait_for_http_server <EOL> def test_external_proxy ( request , io_loop ) : <EOL> """<STR_LIT>""" <EOL> auth_token = '<STR_LIT>' <EOL> proxy_ip = '<STR_LIT:127.0.0.1>' <EOL> proxy_port = <NUM_LIT> <EOL> app = MockHub . instance ( <EOL> proxy_api_ip = proxy_ip , <EOL> proxy_api_port = proxy_port , <EOL> proxy_auth_token = auth_token , <EOL> ) <EOL> def fin ( ) : <EOL> MockHub . clear_instance ( ) <EOL> app . stop ( ) <EOL> request . addfinalizer ( fin ) <EOL> env = os . environ . copy ( ) <EOL> env [ '<STR_LIT>' ] = auth_token <EOL> cmd = app . proxy_cmd + [ <EOL> '<STR_LIT>' , app . ip , <EOL> '<STR_LIT>' , str ( app . port ) , <EOL> '<STR_LIT>' , proxy_ip , <EOL> '<STR_LIT>' , str ( proxy_port ) , <EOL> '<STR_LIT>' , '<STR_LIT>' % ( app . hub_ip , app . hub_port ) , <EOL> ] <EOL> if app . subdomain_host : <EOL> cmd . append ( '<STR_LIT>' ) <EOL> proxy = Popen ( cmd , env = env ) <EOL> def _cleanup_proxy ( ) : <EOL> if proxy . poll ( ) is None : <EOL> proxy . terminate ( ) <EOL> request . addfinalizer ( _cleanup_proxy ) <EOL> def wait_for_proxy ( ) : <EOL> io_loop . run_sync ( lambda : wait_for_http_server ( <EOL> '<STR_LIT>' % ( proxy_ip , proxy_port ) ) ) <EOL> wait_for_proxy ( ) <EOL> app . start ( [ ] ) <EOL> assert app . proxy_process is None <EOL> routes = io_loop . run_sync ( app . proxy . get_routes ) <EOL> assert list ( routes . keys ( ) ) == [ '<STR_LIT:/>' ] <EOL> name = '<STR_LIT>' <EOL> r = api_request ( app , '<STR_LIT>' , name , method = '<STR_LIT>' ) <EOL> r . raise_for_status ( ) <EOL> r = api_request ( app , '<STR_LIT>' , name , '<STR_LIT>' , method = '<STR_LIT>' ) <EOL> r . raise_for_status ( ) <EOL> routes = io_loop . run_sync ( app . proxy . get_routes ) <EOL> user_path = '<STR_LIT>' <EOL> if app . subdomain_host : <EOL> domain = urlparse ( app . subdomain_host ) . hostname <EOL> user_path = '<STR_LIT>' % ( name , domain ) + user_path <EOL> assert sorted ( routes . keys ( ) ) == [ '<STR_LIT:/>' , user_path ] <EOL> proxy . terminate ( ) <EOL> proxy = Popen ( cmd , env = env ) <EOL> wait_for_proxy ( ) <EOL> routes = io_loop . run_sync ( app . proxy . get_routes ) <EOL> assert list ( routes . keys ( ) ) == [ '<STR_LIT:/>' ] <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' ) <EOL> r . raise_for_status ( ) <EOL> routes = io_loop . run_sync ( app . proxy . get_routes ) <EOL> assert sorted ( routes . keys ( ) ) == [ '<STR_LIT:/>' , user_path ] <EOL> proxy . terminate ( ) <EOL> new_auth_token = '<STR_LIT>' <EOL> env [ '<STR_LIT>' ] = new_auth_token <EOL> proxy_port = <NUM_LIT> <EOL> cmd = app . proxy_cmd + [ <EOL> '<STR_LIT>' , app . ip , <EOL> '<STR_LIT>' , str ( app . port ) , <EOL> '<STR_LIT>' , app . proxy_api_ip , <EOL> '<STR_LIT>' , str ( proxy_port ) , <EOL> '<STR_LIT>' , '<STR_LIT>' % ( app . hub_ip , app . hub_port ) , <EOL> ] <EOL> if app . subdomain_host : <EOL> cmd . append ( '<STR_LIT>' ) <EOL> proxy = Popen ( cmd , env = env ) <EOL> wait_for_proxy ( ) <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' , data = json . dumps ( { <EOL> '<STR_LIT:port>' : proxy_port , <EOL> '<STR_LIT>' : '<STR_LIT:http>' , <EOL> '<STR_LIT>' : app . ip , <EOL> '<STR_LIT>' : new_auth_token , <EOL> } ) ) <EOL> r . raise_for_status ( ) <EOL> assert app . proxy . api_server . port == proxy_port <EOL> def get_app_proxy_token ( ) : <EOL> q = Queue ( ) <EOL> app . io_loop . add_callback ( lambda : q . put ( app . proxy . auth_token ) ) <EOL> return q . get ( timeout = <NUM_LIT:2> ) <EOL> assert get_app_proxy_token ( ) == new_auth_token <EOL> app . proxy . auth_token = new_auth_token <EOL> routes = io_loop . run_sync ( app . proxy . get_routes ) <EOL> assert sorted ( routes . keys ( ) ) == [ '<STR_LIT:/>' , user_path ] <EOL> def test_check_routes ( app , io_loop ) : <EOL> proxy = app . proxy <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' ) <EOL> r . raise_for_status ( ) <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' ) <EOL> r . raise_for_status ( ) <EOL> zoe = orm . User . find ( app . db , '<STR_LIT>' ) <EOL> assert zoe is not None <EOL> zoe = app . users [ zoe ] <EOL> before = sorted ( io_loop . run_sync ( app . proxy . get_routes ) ) <EOL> assert zoe . proxy_path in before <EOL> io_loop . run_sync ( lambda : app . proxy . check_routes ( app . users ) ) <EOL> io_loop . run_sync ( lambda : proxy . delete_user ( zoe ) ) <EOL> during = sorted ( io_loop . run_sync ( app . proxy . get_routes ) ) <EOL> assert zoe . proxy_path not in during <EOL> io_loop . run_sync ( lambda : app . proxy . check_routes ( app . users ) ) <EOL> after = sorted ( io_loop . run_sync ( app . proxy . get_routes ) ) <EOL> assert zoe . proxy_path in after <EOL> assert before == after <EOL> def test_patch_proxy_bad_req ( app ) : <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' ) <EOL> assert r . status_code == <NUM_LIT> <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' , data = '<STR_LIT>' ) <EOL> assert r . status_code == <NUM_LIT> <EOL> r = api_request ( app , '<STR_LIT>' , method = '<STR_LIT>' , data = json . dumps ( [ ] ) ) <EOL> assert r . status_code == <NUM_LIT> </s>
