ansible/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2024-09-14 20:13:21 +02:00
Code Releases Activity

Ansible module to enable the user to provision CPGs on HPE StoreServ 3PAR device (#39656)

Browse source 
* Added storage modules and unit tests

Removed unnecessary file

Fixing pep8 issues

Adding reusable documentation to the data fragment file

Fixing issues reported by module validation in documentation

Fixed issues: 10549, 10553

Fixed unit tests

Added future imports and removed license from init py file

Resolved import issue reported by ansible sanity test

Fixed unit tests

Adding CPG management module

Fixing sanity test issues

Resolving line ending issues

Resolved import issue reported by ansible sanity test

Correcting the error message when resource is more than 31 char

Update hpe3par_cpg.py

Fix for defect 10556: Unable to modify volume. Persona not set

Fix for defect 10556: Unable to modify volume. Persona not set

Fixing pep8 changes

Fixed pep8 issues

Added more pep8 related fixes

Added more pep8 related fixes

Fixed rebase issue in cps test

Adding GPL3.0+ License (#7)

* Update hpe3par.py

* Update hpe3par_cpg.py

* Update hpe3par_cpg.py

Adding GPLV3+ only license

Fix anity test

Addressing review comments

Fixing undefined variable error

Addressed review comments and added more unit tests (#8)

Added maintainers

Fixed sanity test

Update test_hpe3par_cpg.py

Renamed the module as per the review comments (#9)

* Added maintainers

* Renaming the module to remove the vendor name from the module

* remove pep8 related issues

Modify the requirements to add version of 3par sdk

Review address (#10)

* Unit is no longer a different paramter in the task

* Fixing unit tests

* Taking into account the default values

Fixing issue with default values

fixing unit test

Adding more flexibility and checks to the size and it's units

Fixed pylint extra line issue

Fixing conflicts in BOTMETA.yml

Made changes according to the review comments

Updating ansible version

Refactored the code to move reusable method to module utils

Modified the GPLv3+ license text according to the review comments. Modified the module utils license to BSD

Corrected the BOTMETA entry. Fixed the pep8 issues

Fixing entries in BOTMETA.yml

Removed ansible version, renamed the version to address the name review comments

Reverting name change for module to revert test failures

Set size and raid type and now reuqired together based on the review comment

Fixed unit tests

Removing default value as raid and set size are now required together

Renaming the modules to ss_3par_*

Resolving the pep8 issues

Added storage modules and unit tests

Removed unnecessary file

Fixing pep8 issues

Adding reusable documentation to the data fragment file

Fixing issues reported by module validation in documentation

Fixed issues: 10549, 10553

Fixed unit tests

Added future imports and removed license from init py file

Resolved import issue reported by ansible sanity test

Fixed unit tests

Adding CPG management module

Fixing sanity test issues

Resolving line ending issues

Resolved import issue reported by ansible sanity test

Correcting the error message when resource is more than 31 char

Update hpe3par_cpg.py

Fix for defect 10556: Unable to modify volume. Persona not set

Fix for defect 10556: Unable to modify volume. Persona not set

Fixing pep8 changes

Fixed pep8 issues

Added more pep8 related fixes

Added more pep8 related fixes

Fixed rebase issue in cps test

Adding GPL3.0+ License (#7)

* Update hpe3par.py

* Update hpe3par_cpg.py

* Update hpe3par_cpg.py

Adding GPLV3+ only license

Fix anity test

Addressing review comments

Fixing undefined variable error

Addressed review comments and added more unit tests (#8)

Added maintainers

Fixed sanity test

Update test_hpe3par_cpg.py

Renamed the module as per the review comments (#9)

* Added maintainers

* Renaming the module to remove the vendor name from the module

* remove pep8 related issues

Modify the requirements to add version of 3par sdk

Review address (#10)

* Unit is no longer a different paramter in the task

* Fixing unit tests

* Taking into account the default values

Fixing issue with default values

fixing unit test

Adding more flexibility and checks to the size and it's units

Fixed pylint extra line issue

Fixing conflicts in BOTMETA.yml

Made changes according to the review comments

Updating ansible version

Refactored the code to move reusable method to module utils

Modified the GPLv3+ license text according to the review comments. Modified the module utils license to BSD

Corrected the BOTMETA entry. Fixed the pep8 issues

Removed ansible version, renamed the version to address the name review comments

Reverting name change for module to revert test failures

Set size and raid type and now reuqired together based on the review comment

Fixed unit tests

Removing default value as raid and set size are now required together

Renaming the modules to ss_3par_*

Resolving the pep8 issues

Update lib/ansible/modules/storage/hpe3par/ss_3par_cpg.py

Co-Authored-By: farhan7500 <farhan7500@gmail.com>

Update lib/ansible/modules/storage/hpe3par/ss_3par_cpg.py

Co-Authored-By: farhan7500 <farhan7500@gmail.com>

Fixed rebase issue

Fixed rebase issuein BOTMETA.yml

* Modified the documentation based on the review comments

* Fixed the pep8 sanity issue

* Moved the doc fragment file based on latest changes

* Reverting inadvertent rebase changes

* Fixed inadvertent change during rebase
This commit is contained in:
Farhan Nomani 2019-02-26 19:39:03 +05:30 committed by John R Barker
parent 076562e5c2
commit 27804b3b8b
8 changed files with 662 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
.github/BOTMETA.yml vendored
View file

@ -394,6 +394,7 @@ files:
labels: gitlab
maintainers: $team_gitlab
notified: jlozadad
$modules/storage/hpe3par/: farhan7500 gautamphegde
$modules/storage/infinidat/: GR360RY vmalloc
$modules/storage/netapp/: $team_netapp
$modules/storage/purestorage/:
@ -782,6 +783,9 @@ files:
$module_utils/vultr.py: *vultr
$module_utils/xenserver.py:
maintainers: bvitnik
$module_utils/storage/hpe3par/hpe3par.py: &hpe3par
maintainers: farhan7500 gautamphegde
support: community
###############################
# playbook
@ -1060,6 +1064,7 @@ files:
$plugins/doc_fragments/vultr.py: *vultr
$plugins/doc_fragments/xenserver.py:
maintainers: bvitnik
$plugins/doc_fragments/hpe3par.py: *hpe3par
###############################
# plugins/filter
$plugins/filter/:
@ -1339,6 +1344,7 @@ files:
labels:
- networking
test/units/modules/cloud/xenserver/: bvitnik
test/units/modules/storage/hpe3par: *hpe3par
test/sanity/pep8/legacy-files.txt:
notified: mattclay
hacking/report.py:

0
lib/ansible/module_utils/storage/hpe3par/__init__.py Normal file
View file

90
lib/ansible/module_utils/storage/hpe3par/hpe3par.py Normal file
View file

@ -0,0 +1,90 @@
# Copyright: (c) 2018, Hewlett Packard Enterprise Development LP
# Simplified BSD License (see licenses/simplified_bsd.txt or https://opensource.org/licenses/BSD-2-Clause)
from ansible.module_utils import basic
def convert_to_binary_multiple(size_with_unit):
if size_with_unit is None:
return -1
valid_units = ['MiB', 'GiB', 'TiB']
valid_unit = False
for unit in valid_units:
if size_with_unit.strip().endswith(unit):
valid_unit = True
size = size_with_unit.split(unit)[0]
if float(size) < 0:
return -1
if not valid_unit:
raise ValueError("%s does not have a valid unit. The unit must be one of %s" % (size_with_unit, valid_units))
size = size_with_unit.replace(" ", "").split('iB')[0]
size_kib = basic.human_to_bytes(size)
return int(size_kib / (1024 * 1024))
storage_system_spec = {
"storage_system_ip": {
"required": True,
"type": "str"
},
"storage_system_username": {
"required": True,
"type": "str",
"no_log": True
},
"storage_system_password": {
"required": True,
"type": "str",
"no_log": True
},
"secure": {
"type": "bool",
"default": False
}
}
def cpg_argument_spec():
spec = {
"state": {
"required": True,
"choices": ['present', 'absent'],
"type": 'str'
},
"cpg_name": {
"required": True,
"type": "str"
},
"domain": {
"type": "str"
},
"growth_increment": {
"type": "str",
},
"growth_limit": {
"type": "str",
},
"growth_warning": {
"type": "str",
},
"raid_type": {
"required": False,
"type": "str",
"choices": ['R0', 'R1', 'R5', 'R6']
},
"set_size": {
"required": False,
"type": "int"
},
"high_availability": {
"type": "str",
"choices": ['PORT', 'CAGE', 'MAG']
},
"disk_type": {
"type": "str",
"choices": ['FC', 'NL', 'SSD']
}
}
spec.update(storage_system_spec)
return spec

0
lib/ansible/modules/storage/hpe3par/__init__.py Normal file
View file

289
lib/ansible/modules/storage/hpe3par/ss_3par_cpg.py Normal file
View file

@ -0,0 +1,289 @@
#!/usr/bin/python
# Copyright: (c) 2018, Hewlett Packard Enterprise Development LP
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import absolute_import, division, print_function
__metaclass__ = type
ANSIBLE_METADATA = {'metadata_version': '1.1',
'status': ['preview'],
'supported_by': 'community'}
DOCUMENTATION = r'''
---
short_description: Manage HPE StoreServ 3PAR CPG
author:
- Farhan Nomani (@farhan7500)
- Gautham P Hegde (@gautamphegde)
description:
- Create and delete CPG on HPE 3PAR.
module: ss_3par_cpg
options:
cpg_name:
description:
- Name of the CPG.
required: true
disk_type:
choices:
- FC
- NL
- SSD
description:
- Specifies that physical disks must have the specified device type.
domain:
description:
- Specifies the name of the domain in which the object will reside.
growth_increment:
description:
- Specifies the growth increment(in MiB, GiB or TiB) the amount of logical disk storage
created on each auto-grow operation.
growth_limit:
description:
- Specifies that the autogrow operation is limited to the specified
storage amount that sets the growth limit(in MiB, GiB or TiB).
growth_warning:
description:
- Specifies that the threshold(in MiB, GiB or TiB) of used logical disk space when exceeded
results in a warning alert.
high_availability:
choices:
- PORT
- CAGE
- MAG
description:
- Specifies that the layout must support the failure of one port pair,
one cage, or one magazine.
raid_type:
choices:
- R0
- R1
- R5
- R6
description:
- Specifies the RAID type for the logical disk.
set_size:
description:
- Specifies the set size in the number of chunklets.
state:
choices:
- present
- absent
description:
- Whether the specified CPG should exist or not.
required: true
secure:
description:
- Specifies whether the certificate needs to be validated while communicating.
type: bool
default: no
extends_documentation_fragment: hpe3par
version_added: 2.8
'''
EXAMPLES = r'''
- name: Create CPG sample_cpg
ss_3par_cpg:
storage_system_ip: 10.10.10.1
storage_system_username: username
storage_system_password: password
state: present
cpg_name: sample_cpg
domain: sample_domain
growth_increment: 32000 MiB
growth_limit: 64000 MiB
growth_warning: 48000 MiB
raid_type: R6
set_size: 8
high_availability: MAG
disk_type: FC
secure: no
- name: Delete CPG sample_cpg
ss_3par_cpg:
storage_system_ip: 10.10.10.1
storage_system_username: username
storage_system_password: password
state: absent
cpg_name: sample_cpg
secure: no
'''
RETURN = r'''
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.storage.hpe3par import hpe3par
try:
from hpe3par_sdk import client
from hpe3parclient import exceptions
HAS_3PARCLIENT = True
except ImportError:
HAS_3PARCLIENT = False
def validate_set_size(raid_type, set_size):
if raid_type:
set_size_array = client.HPE3ParClient.RAID_MAP[raid_type]['set_sizes']
if set_size in set_size_array:
return True
return False
def cpg_ldlayout_map(ldlayout_dict):
if ldlayout_dict['RAIDType'] is not None and ldlayout_dict['RAIDType']:
ldlayout_dict['RAIDType'] = client.HPE3ParClient.RAID_MAP[
ldlayout_dict['RAIDType']]['raid_value']
if ldlayout_dict['HA'] is not None and ldlayout_dict['HA']:
ldlayout_dict['HA'] = getattr(
client.HPE3ParClient, ldlayout_dict['HA'])
return ldlayout_dict
def create_cpg(
client_obj,
cpg_name,
domain,
growth_increment,
growth_limit,
growth_warning,
raid_type,
set_size,
high_availability,
disk_type):
try:
if not validate_set_size(raid_type, set_size):
return (False, False, "Set size %s not part of RAID set %s" % (set_size, raid_type))
if not client_obj.cpgExists(cpg_name):
ld_layout = dict()
disk_patterns = []
if disk_type:
disk_type = getattr(client.HPE3ParClient, disk_type)
disk_patterns = [{'diskType': disk_type}]
ld_layout = {
'RAIDType': raid_type,
'setSize': set_size,
'HA': high_availability,
'diskPatterns': disk_patterns}
ld_layout = cpg_ldlayout_map(ld_layout)
if growth_increment is not None:
growth_increment = hpe3par.convert_to_binary_multiple(
growth_increment)
if growth_limit is not None:
growth_limit = hpe3par.convert_to_binary_multiple(
growth_limit)
if growth_warning is not None:
growth_warning = hpe3par.convert_to_binary_multiple(
growth_warning)
optional = {
'domain': domain,
'growthIncrementMiB': growth_increment,
'growthLimitMiB': growth_limit,
'usedLDWarningAlertMiB': growth_warning,
'LDLayout': ld_layout}
client_obj.createCPG(cpg_name, optional)
else:
return (True, False, "CPG already present")
except exceptions.ClientException as e:
return (False, False, "CPG creation failed | %s" % (e))
return (True, True, "Created CPG %s successfully." % cpg_name)
def delete_cpg(
client_obj,
cpg_name):
try:
if client_obj.cpgExists(cpg_name):
client_obj.deleteCPG(cpg_name)
else:
return (True, False, "CPG does not exist")
except exceptions.ClientException as e:
return (False, False, "CPG delete failed | %s" % e)
return (True, True, "Deleted CPG %s successfully." % cpg_name)
def main():
module = AnsibleModule(argument_spec=hpe3par.cpg_argument_spec(),
required_together=[['raid_type', 'set_size']])
if not HAS_3PARCLIENT:
module.fail_json(msg='the python hpe3par_sdk library is required (https://pypi.org/project/hpe3par_sdk)')
if len(module.params["cpg_name"]) < 1 or len(module.params["cpg_name"]) > 31:
module.fail_json(msg="CPG name must be at least 1 character and not more than 31 characters")
storage_system_ip = module.params["storage_system_ip"]
storage_system_username = module.params["storage_system_username"]
storage_system_password = module.params["storage_system_password"]
cpg_name = module.params["cpg_name"]
domain = module.params["domain"]
growth_increment = module.params["growth_increment"]
growth_limit = module.params["growth_limit"]
growth_warning = module.params["growth_warning"]
raid_type = module.params["raid_type"]
set_size = module.params["set_size"]
high_availability = module.params["high_availability"]
disk_type = module.params["disk_type"]
secure = module.params["secure"]
wsapi_url = 'https://%s:8080/api/v1' % storage_system_ip
try:
client_obj = client.HPE3ParClient(wsapi_url, secure)
except exceptions.SSLCertFailed:
module.fail_json(msg="SSL Certificate Failed")
except exceptions.ConnectionError:
module.fail_json(msg="Connection Error")
except exceptions.UnsupportedVersion:
module.fail_json(msg="Unsupported WSAPI version")
except Exception as e:
module.fail_json(msg="Initializing client failed. %s" % e)
if storage_system_username is None or storage_system_password is None:
module.fail_json(msg="Storage system username or password is None")
if cpg_name is None:
module.fail_json(msg="CPG Name is None")
# States
if module.params["state"] == "present":
try:
client_obj.login(storage_system_username, storage_system_password)
return_status, changed, msg = create_cpg(
client_obj,
cpg_name,
domain,
growth_increment,
growth_limit,
growth_warning,
raid_type,
set_size,
high_availability,
disk_type
)
except Exception as e:
module.fail_json(msg="CPG create failed | %s" % e)
finally:
client_obj.logout()
elif module.params["state"] == "absent":
try:
client_obj.login(storage_system_username, storage_system_password)
return_status, changed, msg = delete_cpg(
client_obj,
cpg_name
)
except Exception as e:
module.fail_json(msg="CPG create failed | %s" % e)
finally:
client_obj.logout()
if return_status:
module.exit_json(changed=changed, msg=msg)
else:
module.fail_json(msg=msg)
if __name__ == '__main__':
main()

29
lib/ansible/plugins/doc_fragments/hpe3par.py Normal file
View file

@ -0,0 +1,29 @@
# Copyright: (c) 2018, Hewlett Packard Enterprise Development LP
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
class ModuleDocFragment(object):
# HPE 3PAR doc fragment
DOCUMENTATION = '''
options:
storage_system_ip:
description:
- The storage system IP address.
required: true
storage_system_password:
description:
- The storage system password.
required: true
storage_system_username:
description:
- The storage system user name.
required: true
requirements:
- hpe3par_sdk >= 1.0.2. Install using 'pip install hpe3par_sdk'
- WSAPI service should be enabled on the 3PAR storage array.
notes:
- check_mode not supported
'''

0
test/units/modules/storage/hpe3par/__init__.py Normal file
View file

248
test/units/modules/storage/hpe3par/test_ss_3par_cpg.py Normal file
View file

@ -0,0 +1,248 @@
# Copyright: (c) 2018, Hewlett Packard Enterprise Development LP
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
import mock
import pytest
import sys
sys.modules['hpe3par_sdk'] = mock.Mock()
sys.modules['hpe3par_sdk.client'] = mock.Mock()
sys.modules['hpe3parclient'] = mock.Mock()
sys.modules['hpe3parclient.exceptions'] = mock.Mock()
from ansible.modules.storage.hpe3par import ss_3par_cpg
from ansible.module_utils.basic import AnsibleModule
from ansible.module_utils.storage.hpe3par import hpe3par
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.AnsibleModule')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.create_cpg')
def test_module_args(mock_create_cpg, mock_module, mock_client):
"""
hpe3par CPG - test module arguments
"""
PARAMS_FOR_PRESENT = {
'storage_system_ip': '192.168.0.1',
'storage_system_username': 'USER',
'storage_system_password': 'PASS',
'cpg_name': 'test_cpg',
'domain': 'test_domain',
'growth_increment': 32768,
'growth_increment_unit': 'MiB',
'growth_limit': 32768,
'growth_limit_unit': 'MiB',
'growth_warning': 32768,
'growth_warning_unit': 'MiB',
'raid_type': 'R6',
'set_size': 8,
'high_availability': 'MAG',
'disk_type': 'FC',
'state': 'present',
'secure': False
}
mock_module.params = PARAMS_FOR_PRESENT
mock_module.return_value = mock_module
mock_client.HPE3ParClient.login.return_value = True
mock_create_cpg.return_value = (True, True, "Created CPG successfully.")
ss_3par_cpg.main()
mock_module.assert_called_with(
argument_spec=hpe3par.cpg_argument_spec(),
required_together=[['raid_type', 'set_size']])
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.AnsibleModule')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.create_cpg')
def test_main_exit_functionality_present_success_without_issue_attr_dict(mock_create_cpg, mock_module, mock_client):
"""
hpe3par flash cache - success check
"""
PARAMS_FOR_PRESENT = {
'storage_system_ip': '192.168.0.1',
'storage_system_name': '3PAR',
'storage_system_username': 'USER',
'storage_system_password': 'PASS',
'cpg_name': 'test_cpg',
'domain': 'test_domain',
'growth_increment': 32768,
'growth_increment_unit': 'MiB',
'growth_limit': 32768,
'growth_limit_unit': 'MiB',
'growth_warning': 32768,
'growth_warning_unit': 'MiB',
'raid_type': 'R6',
'set_size': 8,
'high_availability': 'MAG',
'disk_type': 'FC',
'state': 'present',
'secure': False
}
# This creates a instance of the AnsibleModule mock.
mock_module.params = PARAMS_FOR_PRESENT
mock_module.return_value = mock_module
instance = mock_module.return_value
mock_client.HPE3ParClient.login.return_value = True
mock_create_cpg.return_value = (
True, True, "Created CPG successfully.")
ss_3par_cpg.main()
# AnsibleModule.exit_json should be called
instance.exit_json.assert_called_with(
changed=True, msg="Created CPG successfully.")
# AnsibleModule.fail_json should not be called
assert instance.fail_json.call_count == 0
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.AnsibleModule')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.delete_cpg')
def test_main_exit_functionality_absent_success_without_issue_attr_dict(mock_delete_cpg, mock_module, mock_client):
"""
hpe3par flash cache - success check
"""
PARAMS_FOR_DELETE = {
'storage_system_ip': '192.168.0.1',
'storage_system_name': '3PAR',
'storage_system_username': 'USER',
'storage_system_password': 'PASS',
'cpg_name': 'test_cpg',
'domain': None,
'growth_increment': None,
'growth_increment_unit': None,
'growth_limit': None,
'growth_limit_unit': None,
'growth_warning': None,
'growth_warning_unit': None,
'raid_type': None,
'set_size': None,
'high_availability': None,
'disk_type': None,
'state': 'absent',
'secure': False
}
# This creates a instance of the AnsibleModule mock.
mock_module.params = PARAMS_FOR_DELETE
mock_module.return_value = mock_module
instance = mock_module.return_value
mock_delete_cpg.return_value = (
True, True, "Deleted CPG test_cpg successfully.")
mock_client.HPE3ParClient.login.return_value = True
ss_3par_cpg.main()
# AnsibleModule.exit_json should be called
instance.exit_json.assert_called_with(
changed=True, msg="Deleted CPG test_cpg successfully.")
# AnsibleModule.fail_json should not be called
assert instance.fail_json.call_count == 0
def test_convert_to_binary_multiple():
assert hpe3par.convert_to_binary_multiple(None) == -1
assert hpe3par.convert_to_binary_multiple('-1.0 MiB') == -1
assert hpe3par.convert_to_binary_multiple('-1.0GiB') == -1
assert hpe3par.convert_to_binary_multiple('1.0 MiB') == 1
assert hpe3par.convert_to_binary_multiple('1.5GiB') == 1.5 * 1024
assert hpe3par.convert_to_binary_multiple('1.5 TiB') == 1.5 * 1024 * 1024
assert hpe3par.convert_to_binary_multiple(' 1.5 TiB ') == 1.5 * 1024 * 1024
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
def test_validate_set_size(mock_client):
mock_client.HPE3ParClient.RAID_MAP = {'R0': {'raid_value': 1, 'set_sizes': [1]},
'R1': {'raid_value': 2, 'set_sizes': [2, 3, 4]},
'R5': {'raid_value': 3, 'set_sizes': [3, 4, 5, 6, 7, 8, 9]},
'R6': {'raid_value': 4, 'set_sizes': [6, 8, 10, 12, 16]}
}
raid_type = 'R0'
set_size = 1
assert ss_3par_cpg.validate_set_size(raid_type, set_size)
set_size = 2
assert not ss_3par_cpg.validate_set_size(raid_type, set_size)
raid_type = None
assert not ss_3par_cpg.validate_set_size(raid_type, set_size)
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
def test_cpg_ldlayout_map(mock_client):
mock_client.HPE3ParClient.PORT = 1
mock_client.HPE3ParClient.RAID_MAP = {'R0': {'raid_value': 1, 'set_sizes': [1]},
'R1': {'raid_value': 2, 'set_sizes': [2, 3, 4]},
'R5': {'raid_value': 3, 'set_sizes': [3, 4, 5, 6, 7, 8, 9]},
'R6': {'raid_value': 4, 'set_sizes': [6, 8, 10, 12, 16]}
}
ldlayout_dict = {'RAIDType': 'R6', 'HA': 'PORT'}
assert ss_3par_cpg.cpg_ldlayout_map(ldlayout_dict) == {
'RAIDType': 4, 'HA': 1}
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
def test_create_cpg(mock_client):
ss_3par_cpg.validate_set_size = mock.Mock(return_value=True)
ss_3par_cpg.cpg_ldlayout_map = mock.Mock(
return_value={'RAIDType': 4, 'HA': 1})
mock_client.HPE3ParClient.login.return_value = True
mock_client.HPE3ParClient.cpgExists.return_value = False
mock_client.HPE3ParClient.FC = 1
mock_client.HPE3ParClient.createCPG.return_value = True
assert ss_3par_cpg.create_cpg(mock_client.HPE3ParClient,
'test_cpg',
'test_domain',
'32768 MiB',
'32768 MiB',
'32768 MiB',
'R6',
8,
'MAG',
'FC'
) == (True, True, "Created CPG %s successfully." % 'test_cpg')
mock_client.HPE3ParClient.cpgExists.return_value = True
assert ss_3par_cpg.create_cpg(mock_client.HPE3ParClient,
'test_cpg',
'test_domain',
'32768.0 MiB',
'32768.0 MiB',
'32768.0 MiB',
'R6',
8,
'MAG',
'FC'
) == (True, False, 'CPG already present')
ss_3par_cpg.validate_set_size = mock.Mock(return_value=False)
assert ss_3par_cpg.create_cpg(mock_client.HPE3ParClient,
'test_cpg',
'test_domain',
'32768.0 MiB',
'32768 MiB',
'32768.0 MiB',
'R6',
3,
'MAG',
'FC'
) == (False, False, 'Set size 3 not part of RAID set R6')
@mock.patch('ansible.modules.storage.hpe3par.ss_3par_cpg.client')
def test_delete_cpg(mock_client):
mock_client.HPE3ParClient.login.return_value = True
mock_client.HPE3ParClient.cpgExists.return_value = True
mock_client.HPE3ParClient.FC = 1
mock_client.HPE3ParClient.deleteCPG.return_value = True
assert ss_3par_cpg.delete_cpg(mock_client.HPE3ParClient,
'test_cpg'
) == (True, True, "Deleted CPG %s successfully." % 'test_cpg')
mock_client.HPE3ParClient.cpgExists.return_value = False
assert ss_3par_cpg.delete_cpg(mock_client.HPE3ParClient,
'test_cpg'
) == (True, False, "CPG does not exist")
assert ss_3par_cpg.delete_cpg(mock_client.HPE3ParClient,
None
) == (True, False, "CPG does not exist")