<s> """<STR_LIT>""" <EOL> from uuid import uuid4 <EOL> from fixtures import Fixture <EOL> from characteristic import attributes <EOL> from twisted . python . components import proxyForInterface <EOL> from testtools . matchers import ( <EOL> AllMatch , <EOL> AnyMatch , <EOL> Contains , <EOL> Equals , <EOL> MatchesAll , <EOL> MatchesException , <EOL> MatchesStructure , <EOL> Not , <EOL> Raises , <EOL> ) <EOL> from googleapiclient . errors import HttpError <EOL> from . . blockdevice import ( <EOL> AlreadyAttachedVolume , UnknownVolume , MandatoryProfiles <EOL> ) <EOL> from . . gce import ( <EOL> get_machine_zone , get_machine_project , GCEDiskTypes , GCEVolumeException , <EOL> IGCEOperations <EOL> ) <EOL> from ... . provision . _gce import GCEInstanceBuilder <EOL> from . . testtools import ( <EOL> detach_destroy_volumes , <EOL> get_blockdeviceapi_with_cleanup , <EOL> get_minimum_allocatable_size , <EOL> make_iblockdeviceapi_tests , <EOL> make_icloudapi_tests , <EOL> make_iprofiledblockdeviceapi_tests , <EOL> require_backend , <EOL> ) <EOL> from ... . testtools import TestCase <EOL> @ attributes ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> class GCEComputeTestObjects ( Fixture ) : <EOL> """<STR_LIT>""" <EOL> def _get_instance_builder ( self ) : <EOL> """<STR_LIT>""" <EOL> return GCEInstanceBuilder ( <EOL> compute = self . compute , <EOL> project = self . project , <EOL> zone = self . zone <EOL> ) <EOL> def create_instance ( self , instance_name ) : <EOL> """<STR_LIT>""" <EOL> instance = self . _get_instance_builder ( ) . create_instance ( <EOL> instance_name , <EOL> machine_type = u"<STR_LIT>" <EOL> ) <EOL> def destroy_best_effort ( inst ) : <EOL> try : <EOL> inst . destroy ( ) <EOL> except HttpError as e : <EOL> if e . resp . status == <NUM_LIT> : <EOL> pass <EOL> else : <EOL> raise <EOL> self . addCleanup ( lambda : destroy_best_effort ( instance ) ) <EOL> return instance <EOL> @ require_backend ( '<STR_LIT>' ) <EOL> def gceblockdeviceapi_for_test ( test_case ) : <EOL> """<STR_LIT>""" <EOL> return get_blockdeviceapi_with_cleanup ( test_case ) <EOL> class GCEBlockDeviceAPIInterfaceTests ( <EOL> make_iblockdeviceapi_tests ( <EOL> blockdevice_api_factory = gceblockdeviceapi_for_test , <EOL> unknown_blockdevice_id_factory = lambda test : u"<STR_LIT>" , <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> def test_attach_elsewhere_attached_volume ( self ) : <EOL> pass <EOL> class _RepeatProxy ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , _provider ) : <EOL> """<STR_LIT>""" <EOL> self . _provider = _provider <EOL> def __getattr__ ( self , name ) : <EOL> """<STR_LIT>""" <EOL> method = getattr ( self . _provider , name ) <EOL> def duplicate_proxy ( * args , ** kwargs ) : <EOL> method ( * args , ** kwargs ) <EOL> return method ( * args , ** kwargs ) <EOL> return duplicate_proxy <EOL> def repeat_call_proxy_for ( interface , provider ) : <EOL> """<STR_LIT>""" <EOL> return proxyForInterface ( <EOL> interface , <EOL> originalAttribute = '<STR_LIT>' <EOL> ) ( _RepeatProxy ( _provider = provider ) ) <EOL> class GCEProfiledBlockDeviceApiTests ( <EOL> make_iprofiledblockdeviceapi_tests ( <EOL> profiled_blockdevice_api_factory = gceblockdeviceapi_for_test , <EOL> dataset_size = get_minimum_allocatable_size ( ) ) ) : <EOL> def test_profile_respected ( self ) : <EOL> """<STR_LIT>""" <EOL> for profile in ( c . value for c in MandatoryProfiles . iterconstants ( ) ) : <EOL> dataset_id = uuid4 ( ) <EOL> self . addCleanup ( detach_destroy_volumes , self . api ) <EOL> new_volume = self . api . create_volume_with_profile ( <EOL> dataset_id = dataset_id , <EOL> size = self . dataset_size , <EOL> profile_name = profile <EOL> ) <EOL> if profile in ( MandatoryProfiles . GOLD . value , <EOL> MandatoryProfiles . SILVER . value ) : <EOL> expected_disk_type = GCEDiskTypes . SSD <EOL> else : <EOL> expected_disk_type = GCEDiskTypes . STANDARD <EOL> disk = self . api . _operations . get_disk_details ( <EOL> new_volume . blockdevice_id ) <EOL> actual_disk_type = disk [ '<STR_LIT:type>' ] <EOL> actual_disk_type = actual_disk_type . split ( '<STR_LIT:/>' ) [ - <NUM_LIT:1> ] <EOL> self . assertThat ( <EOL> actual_disk_type , Equals ( expected_disk_type . value ) , <EOL> '<STR_LIT>' . format ( profile ) <EOL> ) <EOL> class GCECloudAPIInterfaceTests ( <EOL> make_icloudapi_tests ( <EOL> blockdevice_api_factory = gceblockdeviceapi_for_test , <EOL> ) <EOL> ) : <EOL> """<STR_LIT>""" <EOL> class GCEBlockDeviceAPITests ( TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_multiple_cluster ( self ) : <EOL> """<STR_LIT>""" <EOL> gce_block_device_api_1 = gceblockdeviceapi_for_test ( self ) <EOL> gce_block_device_api_2 = gceblockdeviceapi_for_test ( self ) <EOL> cluster_1_dataset_id = uuid4 ( ) <EOL> cluster_2_dataset_id = uuid4 ( ) <EOL> gce_block_device_api_1 . create_volume ( cluster_1_dataset_id , <EOL> get_minimum_allocatable_size ( ) ) <EOL> gce_block_device_api_2 . create_volume ( cluster_2_dataset_id , <EOL> get_minimum_allocatable_size ( ) ) <EOL> self . assertEqual ( [ cluster_1_dataset_id ] , <EOL> list ( x . dataset_id <EOL> for x in gce_block_device_api_1 . list_volumes ( ) ) ) <EOL> self . assertEqual ( [ cluster_2_dataset_id ] , <EOL> list ( x . dataset_id <EOL> for x in gce_block_device_api_2 . list_volumes ( ) ) ) <EOL> def test_create_duplicate_dataset_ids ( self ) : <EOL> """<STR_LIT>""" <EOL> gce_block_device_api_1 = gceblockdeviceapi_for_test ( self ) <EOL> gce_block_device_api_2 = gceblockdeviceapi_for_test ( self ) <EOL> shared_dataset_id = uuid4 ( ) <EOL> gce_block_device_api_1 . create_volume ( shared_dataset_id , <EOL> get_minimum_allocatable_size ( ) ) <EOL> self . assertRaises ( <EOL> GCEVolumeException , <EOL> gce_block_device_api_2 . create_volume , <EOL> shared_dataset_id , <EOL> get_minimum_allocatable_size ( ) ) <EOL> def test_list_live_nodes_pagination_and_removal ( self ) : <EOL> """<STR_LIT>""" <EOL> api = gceblockdeviceapi_for_test ( self ) <EOL> api = api . set ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> gce_fixture = self . useFixture ( GCEComputeTestObjects ( <EOL> compute = api . _operations . _compute , <EOL> project = get_machine_project ( ) , <EOL> zone = get_machine_zone ( ) <EOL> ) ) <EOL> other_instance_name = u"<STR_LIT>" + unicode ( uuid4 ( ) ) <EOL> other_instance = gce_fixture . create_instance ( other_instance_name ) <EOL> self . assertThat ( <EOL> api . list_live_nodes ( ) , <EOL> MatchesAll ( <EOL> Contains ( other_instance_name ) , <EOL> Contains ( api . compute_instance_id ( ) ) <EOL> ) <EOL> ) <EOL> api . _stop_node ( other_instance_name ) <EOL> self . assertThat ( <EOL> api . list_live_nodes ( ) , <EOL> MatchesAll ( <EOL> Not ( Contains ( other_instance_name ) ) , <EOL> Contains ( api . compute_instance_id ( ) ) <EOL> ) <EOL> ) <EOL> api . start_node ( other_instance_name ) <EOL> self . assertThat ( <EOL> api . list_live_nodes ( ) , <EOL> MatchesAll ( <EOL> Contains ( other_instance_name ) , <EOL> Contains ( api . compute_instance_id ( ) ) <EOL> ) <EOL> ) <EOL> other_instance . destroy ( ) <EOL> self . assertThat ( <EOL> api . list_live_nodes ( ) , <EOL> MatchesAll ( <EOL> Not ( Contains ( other_instance_name ) ) , <EOL> Contains ( api . compute_instance_id ( ) ) <EOL> ) <EOL> ) <EOL> def test_attach_elsewhere_attached_volume ( self ) : <EOL> """<STR_LIT>""" <EOL> api = gceblockdeviceapi_for_test ( self ) <EOL> gce_fixture = self . useFixture ( GCEComputeTestObjects ( <EOL> compute = api . _operations . _compute , <EOL> project = get_machine_project ( ) , <EOL> zone = get_machine_zone ( ) <EOL> ) ) <EOL> instance_name = u"<STR_LIT>" + unicode ( uuid4 ( ) ) <EOL> other_instance = gce_fixture . create_instance ( instance_name ) <EOL> new_volume = api . create_volume ( <EOL> dataset_id = uuid4 ( ) , <EOL> size = get_minimum_allocatable_size ( ) <EOL> ) <EOL> attached_volume = api . attach_volume ( <EOL> new_volume . blockdevice_id , <EOL> attach_to = other_instance . name , <EOL> ) <EOL> self . assertRaises ( <EOL> AlreadyAttachedVolume , <EOL> api . attach_volume , <EOL> blockdevice_id = attached_volume . blockdevice_id , <EOL> attach_to = api . compute_instance_id ( ) , <EOL> ) <EOL> def test_duplicated_calls ( self ) : <EOL> """<STR_LIT>""" <EOL> actual_api = gceblockdeviceapi_for_test ( self ) <EOL> operations = actual_api . _operations <EOL> api = actual_api . set ( <EOL> '<STR_LIT>' , <EOL> repeat_call_proxy_for ( IGCEOperations , operations ) <EOL> ) <EOL> dataset_id = uuid4 ( ) <EOL> self . assertThat ( <EOL> lambda : api . create_volume ( <EOL> dataset_id = dataset_id , <EOL> size = get_minimum_allocatable_size ( ) <EOL> ) , <EOL> Raises ( MatchesException ( GCEVolumeException ) ) <EOL> ) <EOL> volumes = api . list_volumes ( ) <EOL> self . assertThat ( <EOL> volumes , <EOL> AnyMatch ( MatchesStructure ( dataset_id = Equals ( dataset_id ) ) ) <EOL> ) <EOL> volume = next ( v for v in volumes if v . dataset_id == dataset_id ) <EOL> compute_instance_id = api . compute_instance_id ( ) <EOL> self . assertThat ( <EOL> lambda : api . attach_volume ( <EOL> blockdevice_id = volume . blockdevice_id , <EOL> attach_to = compute_instance_id , <EOL> ) , <EOL> Raises ( MatchesException ( AlreadyAttachedVolume ) ) <EOL> ) <EOL> self . assertThat ( <EOL> api . get_device_path ( volume . blockdevice_id ) . path , <EOL> Contains ( '<STR_LIT>' ) <EOL> ) <EOL> api . detach_volume ( <EOL> blockdevice_id = volume . blockdevice_id , <EOL> ) <EOL> self . assertThat ( <EOL> lambda : api . destroy_volume ( <EOL> blockdevice_id = volume . blockdevice_id , <EOL> ) , <EOL> Raises ( MatchesException ( UnknownVolume ) ) <EOL> ) <EOL> self . assertThat ( <EOL> api . list_volumes ( ) , <EOL> AllMatch ( Not ( MatchesStructure ( dataset_id = Equals ( dataset_id ) ) ) ) <EOL> ) <EOL> def test_list_volumes_walks_pages ( self ) : <EOL> """<STR_LIT>""" <EOL> api = gceblockdeviceapi_for_test ( self ) <EOL> api = api . set ( '<STR_LIT>' , <NUM_LIT:1> ) <EOL> volume_1 = api . create_volume ( <EOL> dataset_id = uuid4 ( ) , <EOL> size = get_minimum_allocatable_size ( ) <EOL> ) <EOL> volume_2 = api . create_volume ( <EOL> dataset_id = uuid4 ( ) , <EOL> size = get_minimum_allocatable_size ( ) <EOL> ) <EOL> blockdevice_ids = [ v . blockdevice_id for v in api . list_volumes ( ) ] <EOL> self . assertThat ( <EOL> blockdevice_ids , <EOL> MatchesAll ( Contains ( volume_1 . blockdevice_id ) , <EOL> Contains ( volume_2 . blockdevice_id ) ) <EOL> ) <EOL> api . destroy_volume ( volume_2 . blockdevice_id ) <EOL> blockdevice_ids = [ v . blockdevice_id for v in api . list_volumes ( ) ] <EOL> self . assertThat ( <EOL> blockdevice_ids , <EOL> MatchesAll ( Contains ( volume_1 . blockdevice_id ) , <EOL> Not ( Contains ( volume_2 . blockdevice_id ) ) ) <EOL> ) </s>
