<s> import sys <EOL> import mock <EOL> import random <EOL> from azure . common import AzureMissingResourceHttpError <EOL> from collections import namedtuple <EOL> from datetime import datetime <EOL> from mock import patch <EOL> from test_helper import * <EOL> from azurectl . account . service import AzureAccount <EOL> from azurectl . azurectl_exceptions import * <EOL> from azurectl . config . parser import Config <EOL> from azurectl . instance . data_disk import DataDisk <EOL> import azurectl <EOL> class TestDataDisk : <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 . data_disk = DataDisk ( 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 . lun = <NUM_LIT:0> <EOL> self . host_caching = '<STR_LIT>' <EOL> self . disk_filename = '<STR_LIT>' <EOL> self . disk_url = ( <EOL> '<STR_LIT>' + <EOL> account . storage_name ( ) + <EOL> '<STR_LIT>' + <EOL> account . get_blob_service_host_base ( ) + '<STR_LIT:/>' + <EOL> account . storage_container ( ) + '<STR_LIT:/>' + <EOL> self . disk_filename <EOL> ) <EOL> self . disk_label = '<STR_LIT>' <EOL> self . disk_size = <NUM_LIT> <EOL> self . timestamp = datetime . utcnow ( ) <EOL> self . time_string = datetime . isoformat ( self . timestamp ) <EOL> def create_mock_data_disk ( self , lun ) : <EOL> return mock . Mock ( <EOL> host_caching = self . host_caching , <EOL> disk_label = self . disk_label , <EOL> disk_name = '<STR_LIT>' , <EOL> lun = lun , <EOL> logical_disk_size_in_gb = self . disk_size , <EOL> media_link = self . disk_url , <EOL> source_media_link = '<STR_LIT>' <EOL> ) <EOL> def create_expected_data_disk_output ( self , lun ) : <EOL> return { <EOL> '<STR_LIT:size>' : '<STR_LIT>' % self . disk_size , <EOL> '<STR_LIT:label>' : self . disk_label , <EOL> '<STR_LIT>' : self . disk_url , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : lun , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> def test_create ( self ) : <EOL> self . service . add_data_disk . return_value = self . my_request <EOL> result = self . data_disk . create ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . disk_size , <EOL> lun = self . lun , <EOL> host_caching = self . host_caching , <EOL> filename = self . disk_filename , <EOL> label = self . disk_label <EOL> ) <EOL> assert result == self . my_request . request_id <EOL> self . service . add_data_disk . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> self . lun , <EOL> host_caching = self . host_caching , <EOL> media_link = self . disk_url , <EOL> disk_label = self . disk_label , <EOL> logical_disk_size_in_gb = self . disk_size <EOL> ) <EOL> @ raises ( AzureDataDiskCreateError ) <EOL> def test_create_upstream_exception ( self ) : <EOL> self . service . add_data_disk . side_effect = Exception <EOL> self . data_disk . create ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . disk_size , <EOL> lun = self . lun , <EOL> host_caching = self . host_caching , <EOL> filename = self . disk_filename , <EOL> label = self . disk_label <EOL> ) <EOL> def test_get_first_available_lun ( self ) : <EOL> self . service . get_data_disk . side_effect = iter ( [ <EOL> self . create_mock_data_disk ( <NUM_LIT:0> ) , <EOL> self . create_mock_data_disk ( <NUM_LIT:1> ) , <EOL> AzureMissingResourceHttpError ( '<STR_LIT>' , <NUM_LIT> ) <EOL> ] ) <EOL> result = self . data_disk . _DataDisk__get_first_available_lun ( <EOL> self . cloud_service_name , <EOL> self . instance_name <EOL> ) <EOL> assert self . service . get_data_disk . call_count == <NUM_LIT:3> <EOL> assert result == <NUM_LIT:2> <EOL> @ raises ( AzureDataDiskNoAvailableLun ) <EOL> def test_no_available_lun_exception ( self ) : <EOL> self . service . get_data_disk . side_effect = iter ( [ <EOL> self . create_mock_data_disk ( i ) for i in range ( <NUM_LIT:16> ) <EOL> ] ) <EOL> self . data_disk . _DataDisk__get_first_available_lun ( <EOL> self . cloud_service_name , <EOL> self . instance_name <EOL> ) <EOL> assert self . service . get_data_disk . call_count == <NUM_LIT:16> <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_without_lun ( self , mock_lun ) : <EOL> self . service . add_data_disk . return_value = self . my_request <EOL> mock_lun . return_value = <NUM_LIT:0> <EOL> result = self . data_disk . create ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . disk_size , <EOL> host_caching = self . host_caching , <EOL> filename = self . disk_filename , <EOL> label = self . disk_label <EOL> ) <EOL> self . service . add_data_disk . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> <NUM_LIT:0> , <EOL> host_caching = self . host_caching , <EOL> media_link = self . disk_url , <EOL> disk_label = self . disk_label , <EOL> logical_disk_size_in_gb = self . disk_size <EOL> ) <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_generate_filename ( self , mock_timestamp ) : <EOL> mock_timestamp . utcnow = mock . Mock ( return_value = self . timestamp ) <EOL> mock_timestamp . isoformat = mock . Mock ( return_value = self . time_string ) <EOL> expected = '<STR_LIT>' % ( <EOL> self . instance_name , <EOL> self . time_string <EOL> ) <EOL> result = self . data_disk . _DataDisk__generate_filename ( self . instance_name ) <EOL> assert result == expected <EOL> @ patch ( '<STR_LIT>' ) <EOL> def test_create_without_filename ( self , mock_generate_filename ) : <EOL> self . service . add_data_disk . return_value = self . my_request <EOL> mock_generate_filename . return_value = self . disk_filename <EOL> result = self . data_disk . create ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . disk_size , <EOL> lun = self . lun , <EOL> host_caching = self . host_caching , <EOL> label = self . disk_label <EOL> ) <EOL> self . service . add_data_disk . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> self . lun , <EOL> host_caching = self . host_caching , <EOL> media_link = self . disk_url , <EOL> disk_label = self . disk_label , <EOL> logical_disk_size_in_gb = self . disk_size <EOL> ) <EOL> def test_show ( self ) : <EOL> self . service . get_data_disk . return_value = self . create_mock_data_disk ( <EOL> self . lun <EOL> ) <EOL> expected = self . create_expected_data_disk_output ( self . lun ) <EOL> result = self . data_disk . show ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . lun <EOL> ) <EOL> self . service . get_data_disk . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> self . lun <EOL> ) <EOL> assert result == expected <EOL> @ raises ( AzureDataDiskShowError ) <EOL> def test_show_upsteam_exception ( self ) : <EOL> self . service . get_data_disk . side_effect = Exception <EOL> self . data_disk . show ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . lun <EOL> ) <EOL> def test_delete ( self ) : <EOL> self . service . delete_data_disk . return_value = self . my_request <EOL> result = self . data_disk . delete ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . lun <EOL> ) <EOL> self . service . delete_data_disk . assert_called_once_with ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . cloud_service_name , <EOL> self . lun , <EOL> delete_vhd = True <EOL> ) <EOL> assert result == self . my_request . request_id <EOL> @ raises ( AzureDataDiskDeleteError ) <EOL> def test_delete_with_upstream_exception ( self ) : <EOL> self . service . delete_data_disk . side_effect = Exception <EOL> self . data_disk . delete ( <EOL> self . cloud_service_name , <EOL> self . instance_name , <EOL> self . lun <EOL> ) <EOL> def test_list ( self ) : <EOL> number_of_disks = random . randint ( <NUM_LIT:0> , <NUM_LIT:15> ) <EOL> luns = random . sample ( range ( <NUM_LIT:16> ) , number_of_disks ) <EOL> luns . sort ( ) <EOL> mock_disk_set = [ AzureMissingResourceHttpError ( '<STR_LIT>' , <NUM_LIT> ) ] * <NUM_LIT:16> <EOL> expected_result = [ ] <EOL> for lun in luns : <EOL> mock_disk_set [ lun ] = self . create_mock_data_disk ( lun ) <EOL> expected_result . append ( self . create_expected_data_disk_output ( lun ) ) <EOL> self . service . get_data_disk . side_effect = iter ( mock_disk_set ) <EOL> result = self . data_disk . list ( <EOL> self . cloud_service_name , <EOL> self . instance_name <EOL> ) <EOL> assert result == expected_result </s>
