<s> """<STR_LIT>""" <EOL> import mock <EOL> from nova . console import websocketproxy <EOL> from nova import exception <EOL> from nova import test <EOL> class NovaProxyRequestHandlerBaseTestCase ( test . NoDBTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( NovaProxyRequestHandlerBaseTestCase , self ) . setUp ( ) <EOL> self . flags ( console_allowed_origins = [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> self . wh = websocketproxy . NovaProxyRequestHandlerBase ( ) <EOL> self . wh . socket = mock . MagicMock ( ) <EOL> self . wh . msg = mock . MagicMock ( ) <EOL> self . wh . do_proxy = mock . MagicMock ( ) <EOL> self . wh . headers = mock . MagicMock ( ) <EOL> def _fake_getheader ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_ipv6 ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_bad_token ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_bad_origin ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_allowed_origin ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_blank_origin ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_no_origin ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return None <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> def _fake_getheader_http ( self , header ) : <EOL> if header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> elif header == '<STR_LIT>' : <EOL> return '<STR_LIT>' <EOL> else : <EOL> return <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_ipv6_url ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_ipv6 <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_token_invalid ( self , check_token ) : <EOL> check_token . return_value = False <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_bad_token <EOL> self . assertRaises ( exception . InvalidToken , <EOL> self . wh . new_websocket_client ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_internal_access_path ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> tsock = mock . MagicMock ( ) <EOL> tsock . recv . return_value = "<STR_LIT>" <EOL> self . wh . socket . return_value = tsock <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( tsock ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_internal_access_path_err ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> tsock = mock . MagicMock ( ) <EOL> tsock . recv . return_value = "<STR_LIT>" <EOL> self . wh . socket . return_value = tsock <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . assertRaises ( exception . InvalidConnectionInfo , <EOL> self . wh . new_websocket_client ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_py273_good_scheme ( <EOL> self , check_token , version_info ) : <EOL> version_info . return_value = ( <NUM_LIT:2> , <NUM_LIT:7> , <NUM_LIT:3> ) <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_py273_special_scheme ( <EOL> self , check_token , version_info ) : <EOL> version_info . return_value = ( <NUM_LIT:2> , <NUM_LIT:7> , <NUM_LIT:3> ) <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . assertRaises ( exception . NovaException , <EOL> self . wh . new_websocket_client ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_address_string_doesnt_do_reverse_dns_lookup ( self , getfqdn ) : <EOL> request_mock = mock . MagicMock ( ) <EOL> request_mock . makefile ( ) . readline . side_effect = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> server_mock = mock . MagicMock ( ) <EOL> client_address = ( '<STR_LIT>' , <NUM_LIT> ) <EOL> handler = websocketproxy . NovaProxyRequestHandler ( <EOL> request_mock , client_address , server_mock ) <EOL> handler . log_message ( '<STR_LIT>' ) <EOL> self . assertFalse ( getfqdn . called ) <EOL> self . assertEqual ( handler . address_string ( ) , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_bad_origin_header ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_bad_origin <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . wh . new_websocket_client ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_allowed_origin_header ( self , <EOL> check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_allowed_origin <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_blank_origin_header ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_blank_origin <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . wh . new_websocket_client ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_no_origin_header ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . socket . return_value = '<STR_LIT>' <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader_no_origin <EOL> self . wh . new_websocket_client ( ) <EOL> check_token . assert_called_with ( mock . ANY , token = "<STR_LIT>" ) <EOL> self . wh . socket . assert_called_with ( '<STR_LIT>' , <NUM_LIT> , connect = True ) <EOL> self . wh . do_proxy . assert_called_with ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_https_origin_proto_http ( self , <EOL> check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . wh . new_websocket_client ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_https_origin_proto_ws ( self , <EOL> check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . wh . new_websocket_client ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_new_websocket_client_novnc_bad_console_type ( self , check_token ) : <EOL> check_token . return_value = { <EOL> '<STR_LIT:host>' : '<STR_LIT>' , <EOL> '<STR_LIT:port>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> self . wh . path = "<STR_LIT>" <EOL> self . wh . headers . getheader = self . _fake_getheader <EOL> self . assertRaises ( exception . ValidationError , <EOL> self . wh . new_websocket_client ) </s>
