<s> import sys <EOL> import mock <EOL> import random <EOL> from collections import namedtuple <EOL> from datetime import datetime <EOL> from mock import patch <EOL> from test_helper import * <EOL> from azure . servicemanagement . models import ( <EOL> PersistentVMRole , <EOL> ConfigurationSets , <EOL> ConfigurationSet , <EOL> ConfigurationSetInputEndpoint , <EOL> DataVirtualHardDisks , <EOL> OSVirtualHardDisk , <EOL> ) <EOL> from azurectl . account . service import AzureAccount <EOL> from azurectl . azurectl_exceptions import * <EOL> from azurectl . config . parser import Config <EOL> from azurectl . instance . endpoint import Endpoint <EOL> import azurectl <EOL> class TestEndpoint : <EOL> def setup ( self ) : <EOL> account = AzureAccount ( <EOL> Config ( <EOL> region_name = '<STR_LIT>' , filename = '<STR_LIT>' <EOL> ) <EOL> ) <EOL> self . service = mock . Mock ( ) <EOL> account . get_management_service = mock . Mock ( return_value = self . service ) <EOL> account . get_blob_service_host_base = mock . Mock ( <EOL> return_value = '<STR_LIT>' <EOL> ) <EOL> account . storage_key = mock . Mock ( ) <EOL> self . endpoint = Endpoint ( account ) <EOL> self . my_request = mock . Mock ( request_id = <NUM_LIT> ) <EOL> self . cloud_service_name = '<STR_LIT>' <EOL> self . instance_name = '<STR_LIT>' <EOL> self . endpoint_name = '<STR_LIT>' <EOL> self . port = '<STR_LIT>' <EOL> self . idle_timeout = <NUM_LIT:10> <EOL> self . protocol = '<STR_LIT>' <EOL> self . udp_endpoint_name = '<STR_LIT>' <EOL> self . udp_port = '<STR_LIT>' <EOL> self . instance_port = '<STR_LIT>' <EOL> self . udp_protocol = '<STR_LIT>' <EOL> self . endpoint . set_instance ( self . cloud_service_name , self . instance_name ) <EOL> self . service . get_role = mock . Mock ( return_value = self . mock_role ( ) ) <EOL> def mock_role ( self , has_endpoint = True ) : <EOL> role = PersistentVMRole ( ) <EOL> role . role_name = self . cloud_service_name <EOL> network_config = ConfigurationSet ( ) <EOL> if has_endpoint : <EOL> endpoint = ConfigurationSetInputEndpoint ( <EOL> name = self . endpoint_name , <EOL> protocol = '<STR_LIT>' , <EOL> port = self . port , <EOL> local_port = self . port , <EOL> idle_timeout_in_minutes = self . idle_timeout <EOL> ) <EOL> network_config . input_endpoints . input_endpoints . append ( endpoint ) <EOL> role . configuration_sets . configuration_sets . append ( network_config ) <EOL> return role <EOL> def create_expected_endpoint_output ( self ) : <EOL> return { <EOL> '<STR_LIT>' : '<STR_LIT>' % self . idle_timeout , <EOL> '<STR_LIT>' : self . port , <EOL> '<STR_LIT:name>' : self . endpoint_name , <EOL> '<STR_LIT:port>' : self . port , <EOL> '<STR_LIT>' : self . protocol <EOL> } <EOL> def test_create ( self ) : <EOL> self . service . update_role . return_value = self . my_request <EOL> mock_role = self . mock_role ( ) <EOL> self . service . get_role = mock . Mock ( return_value = mock_role ) <EOL> endpoint_len = len ( <EOL> mock_role . configuration_sets [ <NUM_LIT:0> ] . input_endpoints <EOL> ) <EOL> result = self . endpoint . create ( <EOL> self . udp_endpoint_name , <EOL> self . udp_port , <EOL> self . instance_port , <EOL> self . udp_protocol , <EOL> self . idle_timeout <EOL> ) <EOL> result_endpoint_length = len ( <EOL> mock_role . configuration_sets [ <NUM_LIT:0> ] . input_endpoints <EOL> ) <EOL> new_endpoint = mock_role . configuration_sets [ <NUM_LIT:0> ] . input_endpoints [ - <NUM_LIT:1> ] <EOL> assert result_endpoint_length == endpoint_len + <NUM_LIT:1> <EOL> assert new_endpoint . name == self . udp_endpoint_name <EOL> assert new_endpoint . port == self . udp_port <EOL> assert new_endpoint . local_port == self . instance_port <EOL> assert new_endpoint . protocol == self . udp_protocol <EOL> self . service . update_role . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> os_virtual_hard_disk = mock_role . os_virtual_hard_disk , <EOL> network_config = mock_role . configuration_sets [ <NUM_LIT:0> ] , <EOL> availability_set_name = mock_role . availability_set_name , <EOL> data_virtual_hard_disks = mock_role . data_virtual_hard_disks <EOL> ) <EOL> assert result == self . my_request . request_id <EOL> @ raises ( AzureEndpointCreateError ) <EOL> def test_create_upstream_exception ( self ) : <EOL> self . service . update_role . side_effect = Exception <EOL> result = self . endpoint . create ( <EOL> self . endpoint_name , <EOL> self . port , <EOL> self . instance_port , <EOL> self . protocol , <EOL> self . idle_timeout <EOL> ) <EOL> def test_show ( self ) : <EOL> expected = self . create_expected_endpoint_output ( ) <EOL> result = self . endpoint . show ( self . endpoint_name ) <EOL> assert result == expected <EOL> @ raises ( AzureEndpointShowError ) <EOL> def test_show_upstream_exception ( self ) : <EOL> self . service . get_role . side_effect = Exception <EOL> self . endpoint . show ( self . endpoint_name ) <EOL> @ raises ( AzureEndpointShowError ) <EOL> def test_show_not_found ( self ) : <EOL> self . service . get_role = mock . Mock ( <EOL> return_value = self . mock_role ( has_endpoint = False ) <EOL> ) <EOL> self . endpoint . show ( self . endpoint_name ) <EOL> def test_list ( self ) : <EOL> expected = [ <EOL> self . create_expected_endpoint_output ( ) <EOL> ] <EOL> result = self . endpoint . list ( ) <EOL> assert result == expected <EOL> @ raises ( AzureEndpointListError ) <EOL> def test_list_upstream_exceptions ( self ) : <EOL> self . service . get_role . side_effect = Exception <EOL> self . endpoint . list ( ) <EOL> def test_delete ( self ) : <EOL> mock_role = self . mock_role ( ) <EOL> self . service . get_role = mock . Mock ( return_value = mock_role ) <EOL> self . service . update_role . return_value = self . my_request <EOL> result = self . endpoint . delete ( '<STR_LIT:foo>' ) <EOL> assert result == self . my_request . request_id <EOL> self . service . update_role . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> os_virtual_hard_disk = mock_role . os_virtual_hard_disk , <EOL> network_config = mock_role . configuration_sets [ <NUM_LIT:0> ] , <EOL> availability_set_name = mock_role . availability_set_name , <EOL> data_virtual_hard_disks = mock_role . data_virtual_hard_disks <EOL> ) <EOL> @ raises ( AzureEndpointDeleteError ) <EOL> def test_delete_with_upstream_exception ( self ) : <EOL> self . service . update_role . side_effect = Exception <EOL> self . endpoint . delete ( self . endpoint_name ) </s>
