diff --git a/.gitattributes b/.gitattributes index 05041c45f..78f5fd340 100644 --- a/.gitattributes +++ b/.gitattributes @@ -1,5 +1,6 @@ # Set the default behavior, in case people don't have core.autocrlf set. * text=auto +*.* text eol=lf # Language aware diff headers # https://tekin.co.uk/2020/10/better-git-diff-output-for-ruby-python-elixir-and-more diff --git a/SoftLayer/CLI/file/cancel.py b/SoftLayer/CLI/file/cancel.py index 9837f9faf..e72898a78 100644 --- a/SoftLayer/CLI/file/cancel.py +++ b/SoftLayer/CLI/file/cancel.py @@ -12,10 +12,8 @@ @click.command(cls=SoftLayer.CLI.command.SLCommand, ) @click.argument('volume-id') @click.option('--reason', help="An optional reason for cancellation") -@click.option('--immediate', - is_flag=True, - help="Cancels the file storage volume immediately instead " - "of on the billing anniversary") +@click.option('--immediate', is_flag=True, + help="Cancels the file storage volume immediately instead of on the billing anniversary") @click.option('--force', default=False, is_flag=True, help="Force modify") @environment.pass_env def cli(env, volume_id, reason, immediate, force): @@ -32,15 +30,12 @@ def cli(env, volume_id, reason, immediate, force): if not (env.skip_confirmations or formatting.no_going_back(volume_id)): raise exceptions.CLIAbort('Aborted.') - cancelled = file_storage_manager.cancel_file_volume(volume_id, - reason, immediate) + cancelled = file_storage_manager.cancel_file_volume(volume_id, reason, immediate) if cancelled: if immediate: - click.echo('File volume with id %s has been marked' - ' for immediate cancellation' % volume_id) + click.echo(f'File volume with id {volume_id} has been marked for immediate cancellation') else: - click.echo('File volume with id %s has been marked' - ' for cancellation' % volume_id) + click.echo(f'File volume with id {volume_id} has been marked for cancellation') else: - click.echo('Unable to cancle file volume %s' % volume_id) + click.echo(f'Unable to cancle file volume {volume_id}') diff --git a/SoftLayer/CLI/formatting.py b/SoftLayer/CLI/formatting.py index fb91ded68..b8c6e4a8f 100644 --- a/SoftLayer/CLI/formatting.py +++ b/SoftLayer/CLI/formatting.py @@ -262,7 +262,7 @@ def no_going_back(confirmation): prompt = f"This action cannot be undone! Type '{confirmation}' or press Enter to abort" ans = click.prompt(prompt, default='', show_default=False) - if ans.lower() == str(confirmation): + if ans.lower() == str(confirmation).lower(): return True return False diff --git a/SoftLayer/managers/storage.py b/SoftLayer/managers/storage.py index 467f03127..310f394be 100644 --- a/SoftLayer/managers/storage.py +++ b/SoftLayer/managers/storage.py @@ -73,10 +73,7 @@ def get_volume_details(self, volume_id, **kwargs): 'notes', ] kwargs['mask'] = ','.join(items) - return self.client.call('Network_Storage', - 'getObject', - id=volume_id, - **kwargs) + return self.client.call('Network_Storage', 'getObject', id=volume_id, **kwargs) def get_volume_access_list(self, volume_id, **kwargs): """Returns a list of authorized hosts for a specified volume. @@ -94,10 +91,7 @@ def get_volume_access_list(self, volume_id, **kwargs): 'allowedIpAddresses[allowedHost[credential]]', ] kwargs['mask'] = ','.join(items) - return self.client.call('Network_Storage', - 'getObject', - id=volume_id, - **kwargs) + return self.client.call('Network_Storage', 'getObject', id=volume_id, **kwargs) def get_volume_snapshot_list(self, volume_id, **kwargs): """Returns a list of snapshots for the specified volume. @@ -115,10 +109,7 @@ def get_volume_snapshot_list(self, volume_id, **kwargs): kwargs['mask'] = ','.join(items) - return self.client.call('Network_Storage', - 'getSnapshots', - id=volume_id, - **kwargs) + return self.client.call('Network_Storage', 'getSnapshots', id=volume_id, **kwargs) def set_volume_snapshot_notification(self, volume_id, enable): """Enables/Disables snapshot space usage threshold warning for a given volume. @@ -144,11 +135,7 @@ def get_volume_snapshot_notification_status(self, volume_id): # instead of either a boolean or real int... return int(status) - def authorize_host_to_volume(self, - volume_id, - hardware_ids=None, - virtual_guest_ids=None, - ip_address_ids=None, + def authorize_host_to_volume(self, volume_id, hardware_ids=None, virtual_guest_ids=None, ip_address_ids=None, subnet_ids=None): """Authorizes hosts to Storage Volumes @@ -163,10 +150,7 @@ def authorize_host_to_volume(self, host_templates = storage_utils.populate_host_templates( hardware_ids, virtual_guest_ids, ip_address_ids, subnet_ids) - return self.client.call('Network_Storage', - 'allowAccessFromHostList', - host_templates, - id=volume_id) + return self.client.call('Network_Storage', 'allowAccessFromHostList', host_templates, id=volume_id) def deauthorize_host_to_volume(self, volume_id, @@ -187,10 +171,7 @@ def deauthorize_host_to_volume(self, host_templates = storage_utils.populate_host_templates( hardware_ids, virtual_guest_ids, ip_address_ids, subnet_ids) - return self.client.call('Network_Storage', - 'removeAccessFromHostList', - host_templates, - id=volume_id) + return self.client.call('Network_Storage', 'removeAccessFromHostList', host_templates, id=volume_id) def get_replication_partners(self, volume_id): """Acquires list of replicant volumes pertaining to the given volume. @@ -198,9 +179,7 @@ def get_replication_partners(self, volume_id): :param volume_id: The ID of the primary volume to be replicated :return: Returns an array of SoftLayer_Location objects """ - return self.client.call('Network_Storage', - 'getReplicationPartners', - id=volume_id) + return self.client.call('Network_Storage', 'getReplicationPartners', id=volume_id) def get_replication_locations(self, volume_id): """Acquires list of the datacenters to which a volume can be replicated. @@ -208,9 +187,7 @@ def get_replication_locations(self, volume_id): :param volume_id: The ID of the primary volume to be replicated :return: Returns an array of SoftLayer_Network_Storage objects """ - return self.client.call('Network_Storage', - 'getValidReplicationTargetDatacenterLocations', - id=volume_id) + return self.client.call('Network_Storage', 'getValidReplicationTargetDatacenterLocations', id=volume_id) def order_replicant_volume(self, volume_id, @@ -249,13 +226,11 @@ def order_replicant_volume(self, if storage_class == 'block': if os_type is None: - if isinstance(utils.lookup(block_volume, 'osType', 'keyName'), - str): + if isinstance(utils.lookup(block_volume, 'osType', 'keyName'), str): os_type = block_volume['osType']['keyName'] else: raise exceptions.SoftLayerError( - "Cannot find primary volume's os-type " - "automatically; must specify manually") + "Cannot find primary volume's os-type automatically; must specify manually") order['osFormatType'] = {'keyName': os_type} return self.client.call('Product_Order', 'placeOrder', order) @@ -285,10 +260,8 @@ def order_duplicate_volume(self, 'storageType[keyName],capacityGb,originalVolumeSize,' \ 'provisionedIops,storageTierLevel,osType[keyName],' \ 'staasVersion,hasEncryptionAtRest' - origin_volume = self.get_volume_details(origin_volume_id, - mask=block_mask) - storage_class = storage_utils.block_or_file( - origin_volume['storageType']['keyName']) + origin_volume = self.get_volume_details(origin_volume_id, mask=block_mask) + storage_class = storage_utils.block_or_file(origin_volume['storageType']['keyName']) order = storage_utils.prepare_duplicate_order_object( self, origin_volume, duplicate_iops, duplicate_tier_level, @@ -296,12 +269,10 @@ def order_duplicate_volume(self, hourly_billing_flag) if storage_class == 'block': - if isinstance(utils.lookup(origin_volume, 'osType', 'keyName'), - str): + if isinstance(utils.lookup(origin_volume, 'osType', 'keyName'), str): os_type = origin_volume['osType']['keyName'] else: - raise exceptions.SoftLayerError( - "Cannot find origin volume's os-type") + raise exceptions.SoftLayerError("Cannot find origin volume's os-type") order['osFormatType'] = {'keyName': os_type} @@ -336,8 +307,7 @@ def order_modified_volume(self, volume_id, new_size=None, new_iops=None, new_tie block_mask = ','.join(mask_items) volume = self.get_volume_details(volume_id, mask=block_mask) - order = storage_utils.prepare_modify_order_object( - self, volume, new_iops, new_tier_level, new_size) + order = storage_utils.prepare_modify_order_object(self, volume, new_iops, new_tier_level, new_size) return self.client.call('Product_Order', 'placeOrder', order) @@ -349,19 +319,14 @@ def volume_set_note(self, volume_id, note): :return: Returns true if success """ template = {'notes': note} - return self.client.call('SoftLayer_Network_Storage', - 'editObject', - template, - id=volume_id) + return self.client.call('SoftLayer_Network_Storage', 'editObject', template, id=volume_id) def delete_snapshot(self, snapshot_id): """Deletes the specified snapshot object. :param snapshot_id: The ID of the snapshot object to delete. """ - return self.client.call('Network_Storage', - 'deleteObject', - id=snapshot_id) + return self.client.call('Network_Storage', 'deleteObject', id=snapshot_id) def create_snapshot(self, volume_id, notes='', **kwargs): """Creates a snapshot on the given block volume. @@ -370,11 +335,7 @@ def create_snapshot(self, volume_id, notes='', **kwargs): :param string notes: The notes or "name" to assign the snapshot :return: Returns the id of the new snapshot """ - return self.client.call('Network_Storage', - 'createSnapshot', - notes, - id=volume_id, - **kwargs) + return self.client.call('Network_Storage', 'createSnapshot', notes, id=volume_id, **kwargs) def order_snapshot_space(self, volume_id, capacity, tier, upgrade, iops=None, **kwargs): @@ -399,10 +360,7 @@ def order_snapshot_space(self, volume_id, capacity, tier, upgrade, return self.client.call('Product_Order', 'placeOrder', order) - def cancel_snapshot_space(self, - volume_id, - reason='No longer needed', - immediate=False): + def cancel_snapshot_space(self, volume_id, reason='No longer needed', immediate=False): """Cancels snapshot space for a given volume. :param integer volume_id: The volume ID @@ -414,8 +372,7 @@ def cancel_snapshot_space(self, volume = self.get_volume_details(volume_id, mask=object_mask) if 'activeChildren' not in volume['billingItem']: - raise exceptions.SoftLayerError( - 'No snapshot space found to cancel') + raise exceptions.SoftLayerError('No snapshot space found to cancel') children_array = volume['billingItem']['activeChildren'] billing_item_id = None @@ -426,18 +383,12 @@ def cancel_snapshot_space(self, break if not billing_item_id: - raise exceptions.SoftLayerError( - 'No snapshot space found to cancel') + raise exceptions.SoftLayerError('No snapshot space found to cancel') if utils.lookup(volume, 'billingItem', 'hourlyFlag'): immediate = True - return self.client.call('SoftLayer_Billing_Item', - 'cancelItem', - immediate, - True, - reason, - id=billing_item_id) + return self.client.call('SoftLayer_Billing_Item', 'cancelItem', immediate, True, reason, id=billing_item_id) def enable_snapshots(self, volume_id, schedule_type, retention_count, minute, hour, day_of_week, **kwargs): @@ -451,15 +402,8 @@ def enable_snapshots(self, volume_id, schedule_type, retention_count, :param string day_of_week: Day when to take snapshot :return: Returns whether successfully scheduled or not """ - return self.client.call('Network_Storage', - 'enableSnapshots', - schedule_type, - retention_count, - minute, - hour, - day_of_week, - id=volume_id, - **kwargs) + return self.client.call('Network_Storage', 'enableSnapshots', schedule_type, retention_count, minute, + hour, day_of_week, id=volume_id, **kwargs) def disable_snapshots(self, volume_id, schedule_type): """Disables snapshots for a specific block volume at a given schedule @@ -468,10 +412,7 @@ def disable_snapshots(self, volume_id, schedule_type): :param string schedule_type: 'HOURLY'|'DAILY'|'WEEKLY' :return: Returns whether successfully disabled or not """ - return self.client.call('Network_Storage', - 'disableSnapshots', - schedule_type, - id=volume_id) + return self.client.call('Network_Storage', 'disableSnapshots', schedule_type, id=volume_id) def list_volume_schedules(self, volume_id): """Lists schedules for a given volume @@ -480,10 +421,7 @@ def list_volume_schedules(self, volume_id): :return: Returns list of schedules assigned to a given volume """ object_mask = 'schedules[type,properties[type]]' - volume_detail = self.client.call('Network_Storage', - 'getObject', - id=volume_id, - mask=object_mask) + volume_detail = self.client.call('Network_Storage', 'getObject', id=volume_id, mask=object_mask) return utils.lookup(volume_detail, 'schedules') @@ -494,10 +432,7 @@ def restore_from_snapshot(self, volume_id, snapshot_id): :param integer snapshot_id: The id of the restore point :return: Returns whether succesfully restored or not """ - return self.client.call('Network_Storage', - 'restoreFromSnapshot', - snapshot_id, - id=volume_id) + return self.client.call('Network_Storage', 'restoreFromSnapshot', snapshot_id, id=volume_id) def failover_to_replicant(self, volume_id, replicant_id): """Failover to a volume replicant. @@ -506,10 +441,7 @@ def failover_to_replicant(self, volume_id, replicant_id): :param integer replicant_id: ID of replicant to failover to :return: Returns whether failover was successful or not """ - return self.client.call('Network_Storage', - 'failoverToReplicant', - replicant_id, - id=volume_id) + return self.client.call('Network_Storage', 'failoverToReplicant', replicant_id, id=volume_id) def disaster_recovery_failover_to_replicant(self, volume_id, replicant_id): """Disaster Recovery Failover to a volume replicant. @@ -518,10 +450,7 @@ def disaster_recovery_failover_to_replicant(self, volume_id, replicant_id): :param integer replicant: ID of replicant to failover to :return: Returns whether failover to successful or not """ - return self.client.call('Network_Storage', - 'disasterRecoveryFailoverToReplicant', - replicant_id, - id=volume_id) + return self.client.call('Network_Storage', 'disasterRecoveryFailoverToReplicant', replicant_id, id=volume_id) def failback_from_replicant(self, volume_id): """Failback from a volume replicant. @@ -529,14 +458,9 @@ def failback_from_replicant(self, volume_id): :param integer volume_id: The id of the volume :return: Returns whether failback was successful or not """ - return self.client.call('Network_Storage', - 'failbackFromReplicant', - id=volume_id) - - def cancel_volume(self, - volume_id, - reason='No longer needed', - immediate=False): + return self.client.call('Network_Storage', 'failbackFromReplicant', id=volume_id) + + def cancel_volume(self, volume_id, reason='No longer needed', immediate=False): """Cancels the given storage volume. :param integer volume_id: The volume ID @@ -547,20 +471,14 @@ def cancel_volume(self, volume = self.get_volume_details(volume_id, mask=object_mask) if 'billingItem' not in volume: - raise exceptions.SoftLayerError( - "Storage Volume was already cancelled") + raise exceptions.SoftLayerError("Storage Volume was already cancelled") billing_item_id = volume['billingItem']['id'] if utils.lookup(volume, 'billingItem', 'hourlyFlag'): immediate = True - return self.client.call('SoftLayer_Billing_Item', - 'cancelItem', - immediate, - True, - reason, - id=billing_item_id) + return self.client.call('SoftLayer_Billing_Item', 'cancelItem', immediate, True, reason, id=billing_item_id) def refresh_dupe(self, volume_id, snapshot_id, force_refresh=False): """"Refresh a duplicate volume with a snapshot from its parent. @@ -569,29 +487,21 @@ def refresh_dupe(self, volume_id, snapshot_id, force_refresh=False): :param integer snapshot_id: The id of the snapshot :param boolean force_refresh: Force refreshing the volume if True """ - return self.client.call('Network_Storage', - 'refreshDuplicate', - snapshot_id, - force_refresh, - id=volume_id) + return self.client.call('Network_Storage', 'refreshDuplicate', snapshot_id, force_refresh, id=volume_id) def convert_dep_dupe(self, volume_id): """Convert a dependent duplicate volume to an independent volume. :param integer volume_id: The id of the volume. """ - return self.client.call('Network_Storage', - 'convertCloneDependentToIndependent', - id=volume_id) + return self.client.call('Network_Storage', 'convertCloneDependentToIndependent', id=volume_id) def convert_dupe_status(self, volume_id): """Get the Clone split/move status completion of a duplicate volume :param integer volume_id: The id of the volume. """ - return self.client.call('Network_Storage', - 'getDuplicateConversionStatus', - id=volume_id) + return self.client.call('Network_Storage', 'getDuplicateConversionStatus', id=volume_id) def get_network_message_delivery_accounts(self, object_id): """Return object data of the cloud storage. @@ -600,8 +510,8 @@ def get_network_message_delivery_accounts(self, object_id): Returns: Get instances """ object_mask = 'mask[uuid,credentials]' - return self.client.call('SoftLayer_Network_Storage_Hub_Cleversafe_Account', - 'getObject', mask=object_mask, id=object_id) + return self.client.call('SoftLayer_Network_Storage_Hub_Cleversafe_Account', 'getObject', + mask=object_mask, id=object_id) def get_end_points(self, object_id): """Returns a collection of endpoint URLs available to this IBM Cloud Object Storage account. @@ -609,5 +519,4 @@ def get_end_points(self, object_id): :param object_id cloud object storage identifier Returns: Returns a collection of endpoint URLs. """ - return self.client.call('SoftLayer_Network_Storage_Hub_Cleversafe_Account', - 'getEndpoints', id=object_id) + return self.client.call('SoftLayer_Network_Storage_Hub_Cleversafe_Account', 'getEndpoints', id=object_id) diff --git a/tests/CLI/helper_tests.py b/tests/CLI/helper_tests.py index 73b727be6..26add91d7 100644 --- a/tests/CLI/helper_tests.py +++ b/tests/CLI/helper_tests.py @@ -29,14 +29,11 @@ def test_default(self): }, cls=formatting.CLIJSONEncoder) self.assertEqual(out, '{"formattedItem": "normal"}') - out = json.dumps({'normal': 'string'}, - cls=formatting.CLIJSONEncoder) + out = json.dumps({'normal': 'string'}, cls=formatting.CLIJSONEncoder) self.assertEqual(out, '{"normal": "string"}') def test_fail(self): - self.assertRaises( - TypeError, - json.dumps, {'test': object()}, cls=formatting.CLIJSONEncoder) + self.assertRaises(TypeError, json.dumps, {'test': object()}, cls=formatting.CLIJSONEncoder) class PromptTests(testing.TestCase): @@ -59,6 +56,12 @@ def test_do_or_die(self, prompt_mock): result = formatting.no_going_back(confirmed) self.assertFalse(result) + # Make sure mixed cases can be checked against + confirmed = "ThisStringHasMixedCase" + prompt_mock.return_value = "thisStringHASMIXEDcase" + result = formatting.no_going_back(confirmed) + self.assertTrue(result) + @mock.patch('click.prompt') def test_confirmation(self, prompt_mock): prompt_mock.return_value = 'Y' @@ -72,16 +75,12 @@ def test_confirmation(self, prompt_mock): prompt_mock.return_value = 'Y' res = formatting.confirm('Confirm?', default=True) self.assertTrue(res) - prompt_mock.assert_called_with('Confirm? [Y/n]', - default='y', - show_default=False) + prompt_mock.assert_called_with('Confirm? [Y/n]', default='y', show_default=False) prompt_mock.return_value = 'N' res = formatting.confirm('Confirm?', default=False) self.assertFalse(res) - prompt_mock.assert_called_with('Confirm? [y/N]', - default='n', - show_default=False) + prompt_mock.assert_called_with('Confirm? [y/N]', default='n', show_default=False) class FormattedItemTests(testing.TestCase): @@ -148,9 +147,7 @@ def test_sort_mixed(self): def test_sort(self): items = [10, formatting.FormattedItem(20), formatting.FormattedItem(5)] sorted_items = sorted(items) - self.assertEqual(sorted_items, [formatting.FormattedItem(5), - 10, - formatting.FormattedItem(20)]) + self.assertEqual(sorted_items, [formatting.FormattedItem(5), 10, formatting.FormattedItem(20)]) class FormattedListTests(testing.TestCase):