5962669522

Committed 24 Aug 2023 10:40AM UTC coverage: 61.185% (-0.2%) from 61.34%

Build # 5962669522

Build Type

Pull #18

github

Committed by

web-flow

Commit Message

Merge c87f581f6 into c9cfcecd1

Pull Request Pull Request #18: Suggesting some new methods

Run Details

34 of 59 new or added lines in 3 files covered. (57.63%)

275 existing lines in 3 files now uncovered.

1146 of 1873 relevant lines covered (61.19%)

2.45 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.38

/pystorcli2/controller/__init__.py

# -*- coding: utf-8 -*-

# Copyright (c) 2018, Martin Dojcak <martin@dojcak.sk>
# Copyright (c) 2022, Rafael Leira & Naudit HPCN S.L. <rafael.leira@naudit.es>
# See LICENSE for details.

'''StorCLI controller python module
'''

from .. import StorCLI
from .. import common
from .. import enclosure
from .. import virtualdrive
from .. import exc
from ..errors import StorcliErrorCode

from datetime import datetime
from typing import List, Optional

# include submodules
from .metrics import ControllerMetrics


class Controller(object):
    """StorCLI Controller

    Instance of this class represents controller in StorCLI hierarchy

    Args:
        ctl_id (str): controller id
        binary (str): storcli binary or full path to the binary

    Properties:
        id (str): controller id
        name (str): controller cmd name
        facts (dict): raw controller facts
        metrics (:obj:ControllerMetrics): controller metrics
        vds (list of :obj:virtualdrive.VirtualDrives): controller virtual drives
        encls (:obj:enclosure.Enclosures): controller enclosures
        autorebuild (dict): current auto rebuild state (also setter)
        foreignautoimport (dict): imports foreign configuration automatically at boot (also setter)
        patrolread (dict): current patrol read settings (also setter)
        cc (dict): current patrol read settings (also setter)
        has_foreign_configurations (bool): true if controller has foreign configurations

    Methods:
        create_vd (:obj:VirtualDrive): create virtual drive
        set_patrolread (dict): configures patrol read state and schedule
        patrolread_start (dict): starts a patrol read on controller
        patrolread_pause (dict): pauses patrol read on controller
        patrolread_resume (dict): resumes patrol read on controller
        patrolread_stop (dict): stops patrol read if running on controller
        patrolread_running (bool): check if patrol read is running on controller
        set_cc (dict): configures consistency check mode and start time
        import_foreign_configurations (dict): imports the foreign configurations on controller
        delete_foreign_configurations (dict): deletes the foreign configuration on controller

    TODO:
        Implement missing methods:
            * patrol read progress
    """

    def __init__(self, ctl_id, binary='storcli64'):
        """Constructor - create StorCLI Controller object

        Args:
            ctl_id (str): controller id
            binary (str): storcli binary or full path to the binary
        """
        self._ctl_id = ctl_id
        self._binary = binary
        self._storcli = StorCLI(binary)
        self._name = '/c{0}'.format(self._ctl_id)

        self._exist()

    def __str__(self):
        return '{0}'.format(common.response_data(self._run(['show'])))

    def _run(self, args, allow_error_codes=[StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION], **kwargs):
        args = args[:]
        args.insert(0, self._name)
        return self._storcli.run(args, allow_error_codes=allow_error_codes, **kwargs)

    def _exist(self):
        try:
            self._run(['show'])
        except exc.StorCliCmdError:
            raise exc.StorCliMissingError(
                self.__class__.__name__, self._name) from None

    @property
    def id(self):
        """ (str): controller id
        """
        return self._ctl_id

    @property
    def name(self):
        """ (str): controller cmd name
        """
        return self._name

    @ property
    def show(self):
        """Static show output to allow getting info of static attributes
        without the need to touch cached responses
        Return:
            dict: controller data from show command
        """
        if not getattr(self, '_show', None):
            out = self._storcli.run(['show'], allow_error_codes=[
                StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION])
            self._show = common.response_data(out)
        return self._show

    @ property
    def serial(self):
        """ (str): get serial number
        """
        return self.show.get('Serial Number', '')

    @ property
    def model(self):
        """ (str): get model
        """
        return self.show.get('Product Name', '')

    @ property
    def pci_address(self):
        """ (str): get pci address
        """
        return self.show.get('PCI Address', '')

    @ property
    def sas_address(self):
        """ (str): get sas address
        """
        return self.show.get('SAS Address', '')

    @property
    def facts(self):
        """ (dict): raw controller facts
        """
        args = [
            'show',
            'all'
        ]
        return common.response_data(self._run(args))

    @property
    def metrics(self):
        """(:obj:ControllerMetrics): controller metrics
        """
        return ControllerMetrics(ctl=self)

    @property
    def vds(self):
        """(:obj:virtualdrive.VirtualDrives): controllers virtual drives
        """
        return virtualdrive.VirtualDrives(ctl_id=self._ctl_id, binary=self._binary)

    @property
    def encls(self):
        """(:obj:enclosure.Enclosures): controller enclosures
        """
        return enclosure.Enclosures(ctl_id=self._ctl_id, binary=self._binary)

    @property
    def drives_ids(self) -> List[str]:
        """(list of str): list of drives ids in format (e:s)
        """
        drives = []
        for encl in self.encls:
            for id in encl.drives.ids:
                drives.append("{enc}:{id}".format(enc=encl.id, id=id))

        return drives

    def create_vd(self, name: str, raid: str, drives: str, strip: str = '64', PDperArray: Optional[int] = None) -> Optional[virtualdrive.VirtualDrive]:
        """Create virtual drive (raid) managed by current controller

        Args:
            name (str): virtual drive name
            raid (str): virtual drive raid level (raid0, raid1, ...)
            drives (str): storcli drives expression (e:s|e:s-x|e:s-x,y;e:s-x,y,z)
            strip (str, optional): virtual drive raid strip size

        Returns:
            (None): no virtual drive created with name
            (:obj:virtualdrive.VirtualDrive)
        """
        args = [
            'add',
            'vd',
            'r{0}'.format(raid),
            'name={0}'.format(name),
            'drives={0}'.format(drives),
            'strip={0}'.format(strip)
        ]

        try:
            if int(raid) >= 10 and PDperArray is None:
                # Try to count the number of drives in the array
                # The format of the drives argument is e:s|e:s-x|e:s-x,y;e:s-x,y,z

                numDrives = common.count_drives(drives)

                if numDrives % 2 != 0 and numDrives % 3 == 0:
                    # In some scenarios, such as 9 drives with raid 60, 3 is a good pd number but 4 is not
                    # Must check for similar scenarios
                    # BTW we don't clearly understand what PDperArray is for and what exactly it does under the hood. More investigation is needed
                    PDperArray = numDrives//3
                else:
                    PDperArray = numDrives//2

        except ValueError:
            pass

        finally:
            if raid == '00' and PDperArray is None:
                PDperArray = 1

        if PDperArray is not None:
            args.append('PDperArray={0}'.format(PDperArray))

        self._run(args)
        for vd in self.vds:
            if name == vd.name:
                return vd
        return None

    @property
    @common.lower
    def autorebuild(self):
        """Get/Set auto rebuild state

        One of the following options can be set (str):
            on - enables autorebuild
            off - disables autorebuild

        Returns:
            (str): on / off
        """
        args = [
            'show',
            'autorebuild'
        ]

        prop = common.response_property(self._run(args))[0]
        return prop['Value']

    @autorebuild.setter
    def autorebuild(self, value):
        """
        """
        args = [
            'set',
            'autorebuild={0}'.format(value)
        ]
        return common.response_setter(self._run(args))

    @property
    @common.lower
    def foreignautoimport(self):
        """Get/Set auto foreign import configuration

        One of the following options can be set (str):
            on - enables foreignautoimport
            off - disables foreignautoimport

        Returns:
            (str): on / off
        """
        args = [
            'show',
            'foreignautoimport'
        ]
        prop = common.response_property(self._run(args))[0]
        return prop['Value']

    @foreignautoimport.setter
    def foreignautoimport(self, value):
        """
        """
        args = [
            'set',
            'foreignautoimport={0}'.format(value)
        ]
        return common.response_setter(self._run(args))

    @property
    @common.lower
    def patrolread(self):
        """Get/Set patrol read

        One of the following options can be set (str):
            on - enables patrol read
            off - disables patrol read

        Returns:
            (str): on / off
        """
        args = [
            'show',
            'patrolread'
        ]

        for pr in common.response_property(self._run(args)):
            if pr['Ctrl_Prop'] == "PR Mode":
                if pr['Value'] == 'Disable':
                    return 'off'
                else:
                    return 'on'
        return 'off'

    @patrolread.setter
    def patrolread(self, value):
        """
        """
        return self.set_patrolread(value)

    def set_patrolread(self, value, mode='manual'):
        """Set patrol read

        Args:
            value (str): on / off to configure patrol read state
            mode (str): auto | manual to configure patrol read schedule
        """
        args = [
            'set',
            'patrolread={0}'.format(value)
        ]

        if value == 'on':
            args.append('mode={0}'.format(mode))

        return common.response_setter(self._run(args))

    def patrolread_start(self):
        """Starts the patrol read operation of the controller

        Returns:
            (dict): response cmd data
        """
        args = [
            'start',
            'patrolread'
        ]
        return common.response_cmd(self._run(args))

    def patrolread_stop(self):
        """Stops the patrol read operation of the controller

        Returns:
            (dict): response cmd data
        """
        args = [
            'stop',
            'patrolread'
        ]
        return common.response_cmd(self._run(args))

    def patrolread_pause(self):
        """Pauses the patrol read operation of the controller

        Returns:
            (dict): response cmd data
        """
        args = [
            'pause',
            'patrolread'
        ]
        return common.response_cmd(self._run(args))

    def patrolread_resume(self):
        """Resumes the patrol read operation of the controller

        Returns:
            (dict): response cmd data
        """
        args = [
            'resume',
            'patrolread'
        ]
        return common.response_cmd(self._run(args))

    @property
    def patrolread_running(self):
        """Check if patrol read is running on the controller

        Returns:
            (bool): true / false
        """
        args = [
            'show',
            'patrolread'
        ]

        status = ''
        for pr in common.response_property(self._run(args)):
            if pr['Ctrl_Prop'] == "PR Current State":
                status = pr['Value']
        return bool('Active' in status)

    @property
    @common.lower
    def cc(self):
        """Get/Set consistency chceck

        One of the following options can be set (str):
            seq  - sequential mode
            conc - concurrent mode
            off  - disables consistency check

        Returns:
            (str): on / off
        """
        args = [
            'show',
            'cc'
        ]

        for pr in common.response_property(self._run(args)):
            if pr['Ctrl_Prop'] == "CC Operation Mode":
                if pr['Value'] == 'Disabled':
                    return 'off'
                else:
                    return 'on'
        return 'off'

    @cc.setter
    def cc(self, value):
        """
        """
        return self.set_cc(value)

    def set_cc(self, value, starttime=None):
        """Set consistency check

        Args:
            value (str):
                seq  - sequential mode
                conc - concurrent mode
                off  - disables consistency check
            starttime (str): Start time of a consistency check is yyyy/mm/dd hh format
        """
        args = [
            'set',
            'cc={0}'.format(value)
        ]

        if value in ('seq', 'conc'):
            if starttime is None:
                starttime = datetime.now().strftime('%Y/%m/%d %H')
            args.append('starttime="{0}"'.format(starttime))

        return common.response_setter(self._run(args))

    def has_foreign_configurations(self, securitykey: Optional[str] = None) -> bool:
        """(bool): true if controller has foreign configurations
        """
        args = [
            '/fall',
            'show'
        ]

        if securitykey:
            args.append(f'securitykey={securitykey}')

        try:
            fc_data = common.response_data(self._run(args))
            fcs = 0

            if 'Total foreign Drive Groups' in fc_data:
                fcs = int(fc_data['Total foreign Drive Groups'])
            if 'Total Foreign PDs' in fc_data:
                fcs += int(fc_data['Total Foreign PDs'])
            if 'Total Locked Foreign PDs' in fc_data:
                fcs += int(fc_data['Total Locked Foreign PDs'])

            if fcs > 0:
                return True
        except KeyError:
            pass
        return False

    def is_foreign_configuration_healthy(self, securitykey: Optional[str] = None) -> bool:
        """(bool): true if controller has healthy foreign configurations
        """

        if not self.has_foreign_configurations(securitykey):
            return True

        args = [
            '/fall',
            'show'
        ]

        if securitykey:
            args.append(f'securitykey={securitykey}')

        try:
            fc_data = common.response_data(
                self._run(args, allow_error_codes=[]))
        except exc.StorCliCmdErrorCode as e:
            if e.error_code == StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION:
                return False

            raise e

        return True

    def delete_foreign_configurations(self, securitykey: Optional[str] = None):
        """Deletes foreign configurations

        Returns:
            (dict): response cmd data
        """
        args = [
            '/fall',
            'del'
        ]

        if securitykey:
            args.append(f'securitykey={securitykey}')
        return common.response_cmd(self._run(args))

    def import_foreign_configurations(self, securitykey: Optional[str] = None):
        """Imports foreign configurations

        Returns:
            (dict): response cmd data
        """
        args = [
            '/fall',
            'import'
        ]
        if securitykey:
            args.append(f'securitykey={securitykey}')
        return common.response_cmd(self._run(args))

    def set_controller_mode(self, mode):
        """Set controller mode command

        Returns:
            (dict): response cmd data
        """
        args = [
            'set',
            f'personality={mode}'
        ]
        return common.response_cmd(self._run(args))

    def delete_vds(self):
        """Delete all virtual drives

        Returns:
            (dict): response cmd data
        """
        args = [
            '/vall',
            'delete'
        ]
        return common.response_cmd(self._run(args))

    @property
    def phyerrorcounters(self):
        """Get Phy Error Counters

        Returns:
            (dict): response cmd data
        """
        args = [
            '/pall',
            'show'
        ]
        try:
            # RAID
            return common.response_data(self._run(args + ['all']))['Phy Error Counters']
        except exc.StorCliCmdError:
            # HBA
            return common.response_data(self._run(args + ['phyerrorcounters']))


class Controllers(object):
    """StorCLI Controllers

    Instance of this class is iterable with :obj:Controller as item

    Args:
        binary (str): storcli binary or full path to the binary

    Properties:
        ids (list of str): list of controllers id

    Methods:
        get_clt (:obj:Controller): return controller object by id
    """

    def __init__(self, binary='storcli64'):
        """Constructor - create StorCLI Controllers object

        Args:
            binary (str): storcli binary or full path to the binary
        """
        self._binary = binary
        self._storcli = StorCLI(binary)

    @ property
    def show(self) -> List[dict]:
        """Static show output to allow getting info of static attributes
        without the need to touch cached responses
        Return:
            list: list of dicts of controllers data from show command
        """
        if not getattr(self, '_show', None):
            out = self._storcli.run(['show'], allow_error_codes=[
                StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION])
            response = common.response_data(out)
            if "Number of Controllers" in response and response["Number of Controllers"] == 0:
                self._show = []
            else:
                self._show = common.response_data_subkey(
                    out, ['System Overview', 'IT System Overview'])
        return self._show

    @ property
    def _ctl_ids(self) -> List[int]:
        out = self.show
        return [] if not out else [ctl['Ctl'] for ctl in out]

    @ property
    def _ctls(self):
        for ctl_id in self._ctl_ids:
            yield Controller(ctl_id=ctl_id, binary=self._binary)

    def __iter__(self):
        return self._ctls

    @ property
    def ids(self):
        """(list of str): controllers id
        """
        return self._ctl_ids

    def get_ctl(self, ctl_id: int) -> Optional[Controller]:
        """Get controller object by id

        Args:
            ctl_id (str): controller id

        Returns:
            (None): no controller with id
            (:obj:Controller): controller object
        """
        for ctl in self:
            if ctl.id == ctl_id:
                return ctl
        return None

1	# -- coding: utf-8 --
2
3	# Copyright (c) 2018, Martin Dojcak <martin@dojcak.sk>
4	# Copyright (c) 2022, Rafael Leira & Naudit HPCN S.L. <rafael.leira@naudit.es>
5	# See LICENSE for details.
6
7	'''StorCLI controller python module	4✔
8	'''
9
10	from .. import StorCLI	4✔
11	from .. import common	4✔
12	from .. import enclosure	4✔
13	from .. import virtualdrive	4✔
14	from .. import exc	4✔
15	from ..errors import StorcliErrorCode	4✔
16
17	from datetime import datetime	4✔
18	from typing import List, Optional	4✔
19
20	# include submodules
21	from .metrics import ControllerMetrics	4✔
22
23
24	class Controller(object):	4✔
25	"""StorCLI Controller	4✔
26
27	Instance of this class represents controller in StorCLI hierarchy
28
29	Args:
30	ctl_id (str): controller id
31	binary (str): storcli binary or full path to the binary
32
33	Properties:
34	id (str): controller id
35	name (str): controller cmd name
36	facts (dict): raw controller facts
37	metrics (:obj:ControllerMetrics): controller metrics
38	vds (list of :obj:virtualdrive.VirtualDrives): controller virtual drives
39	encls (:obj:enclosure.Enclosures): controller enclosures
40	autorebuild (dict): current auto rebuild state (also setter)
41	foreignautoimport (dict): imports foreign configuration automatically at boot (also setter)
42	patrolread (dict): current patrol read settings (also setter)
43	cc (dict): current patrol read settings (also setter)
44	has_foreign_configurations (bool): true if controller has foreign configurations
45
46	Methods:
47	create_vd (:obj:VirtualDrive): create virtual drive
48	set_patrolread (dict): configures patrol read state and schedule
49	patrolread_start (dict): starts a patrol read on controller
50	patrolread_pause (dict): pauses patrol read on controller
51	patrolread_resume (dict): resumes patrol read on controller
52	patrolread_stop (dict): stops patrol read if running on controller
53	patrolread_running (bool): check if patrol read is running on controller
54	set_cc (dict): configures consistency check mode and start time
55	import_foreign_configurations (dict): imports the foreign configurations on controller
56	delete_foreign_configurations (dict): deletes the foreign configuration on controller
57
58	TODO:
59	Implement missing methods:
60	* patrol read progress
61	"""
62
63	def __init__(self, ctl_id, binary='storcli64'):	4✔
64	"""Constructor - create StorCLI Controller object
65
66	Args:
67	ctl_id (str): controller id
68	binary (str): storcli binary or full path to the binary
69	"""
70	self._ctl_id = ctl_id	4✔
71	self._binary = binary	4✔
72	self._storcli = StorCLI(binary)	4✔
73	self._name = '/c{0}'.format(self._ctl_id)	4✔
74
75	self._exist()	4✔
76
77	def __str__(self):	4✔
78	return '{0}'.format(common.response_data(self._run(['show'])))	×
79
80	def _run(self, args, allow_error_codes=[StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION], **kwargs):	4✔
81	args = args[:]	4✔
82	args.insert(0, self._name)	4✔
83	return self._storcli.run(args, allow_error_codes=allow_error_codes, **kwargs)	4✔
84
85	def _exist(self):	4✔
86	try:	4✔
87	self._run(['show'])	4✔
88	except exc.StorCliCmdError:	×
89	raise exc.StorCliMissingError(	×
90	self.__class__.__name__, self._name) from None
91
92	@property	4✔
93	def id(self):	4✔
94	""" (str): controller id
95	"""
96	return self._ctl_id	4✔
97
98	@property	4✔
99	def name(self):	4✔
100	""" (str): controller cmd name
101	"""
102	return self._name	×
103
104	@ property	4✔
105	def show(self):	4✔
106	"""Static show output to allow getting info of static attributes
107	without the need to touch cached responses
108	Return:
109	dict: controller data from show command
110	"""
NEW 111	if not getattr(self, '_show', None):	×
NEW 112	out = self._storcli.run(['show'], allow_error_codes=[	×
113	StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION])
NEW 114	self._show = common.response_data(out)	×
NEW 115	return self._show	×
116
117	@ property	4✔
118	def serial(self):	4✔
119	""" (str): get serial number
120	"""
NEW 121	return self.show.get('Serial Number', '')	×
122
123	@ property	4✔
124	def model(self):	4✔
125	""" (str): get model
126	"""
NEW 127	return self.show.get('Product Name', '')	×
128
129	@ property	4✔
130	def pci_address(self):	4✔
131	""" (str): get pci address
132	"""
NEW 133	return self.show.get('PCI Address', '')	×
134
135	@ property	4✔
136	def sas_address(self):	4✔
137	""" (str): get sas address
138	"""
NEW UNCOV 139	return self.show.get('SAS Address', '')	×
140
141	@property	4✔
142	def facts(self):	4✔
143	""" (dict): raw controller facts
144	"""
UNCOV 145	args = [	×
146	'show',
147	'all'
148	]
UNCOV 149	return common.response_data(self._run(args))	×
150
151	@property	4✔
152	def metrics(self):	4✔
153	"""(:obj:ControllerMetrics): controller metrics
154	"""
UNCOV 155	return ControllerMetrics(ctl=self)	×
156
157	@property	4✔
158	def vds(self):	4✔
159	"""(:obj:virtualdrive.VirtualDrives): controllers virtual drives
160	"""
161	return virtualdrive.VirtualDrives(ctl_id=self._ctl_id, binary=self._binary)	4✔
162
163	@property	4✔
164	def encls(self):	4✔
165	"""(:obj:enclosure.Enclosures): controller enclosures
166	"""
167	return enclosure.Enclosures(ctl_id=self._ctl_id, binary=self._binary)	4✔
168
169	@property	4✔
170	def drives_ids(self) -> List[str]:	4✔
171	"""(list of str): list of drives ids in format (e:s)
172	"""
UNCOV 173	drives = []	×
UNCOV 174	for encl in self.encls:	×
UNCOV 175	for id in encl.drives.ids:	×
UNCOV 176	drives.append("{enc}:{id}".format(enc=encl.id, id=id))	×
177
UNCOV 178	return drives	×
179
180	def create_vd(self, name: str, raid: str, drives: str, strip: str = '64', PDperArray: Optional[int] = None) -> Optional[virtualdrive.VirtualDrive]:	4✔
181	"""Create virtual drive (raid) managed by current controller
182
183	Args:
184	name (str): virtual drive name
185	raid (str): virtual drive raid level (raid0, raid1, ...)
186	drives (str): storcli drives expression (e:s\|e:s-x\|e:s-x,y;e:s-x,y,z)
187	strip (str, optional): virtual drive raid strip size
188
189	Returns:
190	(None): no virtual drive created with name
191	(:obj:virtualdrive.VirtualDrive)
192	"""
193	args = [	4✔
194	'add',
195	'vd',
196	'r{0}'.format(raid),
197	'name={0}'.format(name),
198	'drives={0}'.format(drives),
199	'strip={0}'.format(strip)
200	]
201
202	try:	4✔
203	if int(raid) >= 10 and PDperArray is None:	4✔
204	# Try to count the number of drives in the array
205	# The format of the drives argument is e:s\|e:s-x\|e:s-x,y;e:s-x,y,z
206
UNCOV 207	numDrives = common.count_drives(drives)	×
208
UNCOV 209	if numDrives % 2 != 0 and numDrives % 3 == 0:	×
210	# In some scenarios, such as 9 drives with raid 60, 3 is a good pd number but 4 is not
211	# Must check for similar scenarios
212	# BTW we don't clearly understand what PDperArray is for and what exactly it does under the hood. More investigation is needed
UNCOV 213	PDperArray = numDrives//3	×
214	else:
215	PDperArray = numDrives//2	×
216
UNCOV 217	except ValueError:	×
218	pass	×
219
220	finally:
221	if raid == '00' and PDperArray is None:	4✔
UNCOV 222	PDperArray = 1	×
223
224	if PDperArray is not None:	4✔
UNCOV 225	args.append('PDperArray={0}'.format(PDperArray))	×
226
227	self._run(args)	4✔
228	for vd in self.vds:	4✔
229	if name == vd.name:	4✔
230	return vd	4✔
UNCOV 231	return None	×
232
233	@property	4✔
234	@common.lower	4✔
235	def autorebuild(self):	4✔
236	"""Get/Set auto rebuild state
237
238	One of the following options can be set (str):
239	on - enables autorebuild
240	off - disables autorebuild
241
242	Returns:
243	(str): on / off
244	"""
UNCOV 245	args = [	×
246	'show',
247	'autorebuild'
248	]
249
250	prop = common.response_property(self._run(args))[0]	×
UNCOV 251	return prop['Value']	×
252
253	@autorebuild.setter	4✔
254	def autorebuild(self, value):	4✔
255	"""
256	"""
UNCOV 257	args = [	×
258	'set',
259	'autorebuild={0}'.format(value)
260	]
UNCOV 261	return common.response_setter(self._run(args))	×
262
263	@property	4✔
264	@common.lower	4✔
265	def foreignautoimport(self):	4✔
266	"""Get/Set auto foreign import configuration
267
268	One of the following options can be set (str):
269	on - enables foreignautoimport
270	off - disables foreignautoimport
271
272	Returns:
273	(str): on / off
274	"""
UNCOV 275	args = [	×
276	'show',
277	'foreignautoimport'
278	]
279	prop = common.response_property(self._run(args))[0]	×
UNCOV 280	return prop['Value']	×
281
282	@foreignautoimport.setter	4✔
283	def foreignautoimport(self, value):	4✔
284	"""
285	"""
UNCOV 286	args = [	×
287	'set',
288	'foreignautoimport={0}'.format(value)
289	]
UNCOV 290	return common.response_setter(self._run(args))	×
291
292	@property	4✔
293	@common.lower	4✔
294	def patrolread(self):	4✔
295	"""Get/Set patrol read
296
297	One of the following options can be set (str):
298	on - enables patrol read
299	off - disables patrol read
300
301	Returns:
302	(str): on / off
303	"""
304	args = [	×
305	'show',
306	'patrolread'
307	]
308
UNCOV 309	for pr in common.response_property(self._run(args)):	×
UNCOV 310	if pr['Ctrl_Prop'] == "PR Mode":	×
UNCOV 311	if pr['Value'] == 'Disable':	×
UNCOV 312	return 'off'	×
313	else:
314	return 'on'	×
UNCOV 315	return 'off'	×
316
317	@patrolread.setter	4✔
318	def patrolread(self, value):	4✔
319	"""
320	"""
UNCOV 321	return self.set_patrolread(value)	×
322
323	def set_patrolread(self, value, mode='manual'):	4✔
324	"""Set patrol read
325
326	Args:
327	value (str): on / off to configure patrol read state
328	mode (str): auto \| manual to configure patrol read schedule
329	"""
UNCOV 330	args = [	×
331	'set',
332	'patrolread={0}'.format(value)
333	]
334
UNCOV 335	if value == 'on':	×
UNCOV 336	args.append('mode={0}'.format(mode))	×
337
UNCOV 338	return common.response_setter(self._run(args))	×
339
340	def patrolread_start(self):	4✔
341	"""Starts the patrol read operation of the controller
342
343	Returns:
344	(dict): response cmd data
345	"""
UNCOV 346	args = [	×
347	'start',
348	'patrolread'
349	]
UNCOV 350	return common.response_cmd(self._run(args))	×
351
352	def patrolread_stop(self):	4✔
353	"""Stops the patrol read operation of the controller
354
355	Returns:
356	(dict): response cmd data
357	"""
UNCOV 358	args = [	×
359	'stop',
360	'patrolread'
361	]
UNCOV 362	return common.response_cmd(self._run(args))	×
363
364	def patrolread_pause(self):	4✔
365	"""Pauses the patrol read operation of the controller
366
367	Returns:
368	(dict): response cmd data
369	"""
UNCOV 370	args = [	×
371	'pause',
372	'patrolread'
373	]
UNCOV 374	return common.response_cmd(self._run(args))	×
375
376	def patrolread_resume(self):	4✔
377	"""Resumes the patrol read operation of the controller
378
379	Returns:
380	(dict): response cmd data
381	"""
UNCOV 382	args = [	×
383	'resume',
384	'patrolread'
385	]
UNCOV 386	return common.response_cmd(self._run(args))	×
387
388	@property	4✔
389	def patrolread_running(self):	4✔
390	"""Check if patrol read is running on the controller
391
392	Returns:
393	(bool): true / false
394	"""
395	args = [	×
396	'show',
397	'patrolread'
398	]
399
UNCOV 400	status = ''	×
UNCOV 401	for pr in common.response_property(self._run(args)):	×
UNCOV 402	if pr['Ctrl_Prop'] == "PR Current State":	×
UNCOV 403	status = pr['Value']	×
UNCOV 404	return bool('Active' in status)	×
405
406	@property	4✔
407	@common.lower	4✔
408	def cc(self):	4✔
409	"""Get/Set consistency chceck
410
411	One of the following options can be set (str):
412	seq - sequential mode
413	conc - concurrent mode
414	off - disables consistency check
415
416	Returns:
417	(str): on / off
418	"""
419	args = [	×
420	'show',
421	'cc'
422	]
423
UNCOV 424	for pr in common.response_property(self._run(args)):	×
UNCOV 425	if pr['Ctrl_Prop'] == "CC Operation Mode":	×
UNCOV 426	if pr['Value'] == 'Disabled':	×
UNCOV 427	return 'off'	×
428	else:
429	return 'on'	×
UNCOV 430	return 'off'	×
431
432	@cc.setter	4✔
433	def cc(self, value):	4✔
434	"""
435	"""
UNCOV 436	return self.set_cc(value)	×
437
438	def set_cc(self, value, starttime=None):	4✔
439	"""Set consistency check
440
441	Args:
442	value (str):
443	seq - sequential mode
444	conc - concurrent mode
445	off - disables consistency check
446	starttime (str): Start time of a consistency check is yyyy/mm/dd hh format
447	"""
448	args = [	×
449	'set',
450	'cc={0}'.format(value)
451	]
452
UNCOV 453	if value in ('seq', 'conc'):	×
UNCOV 454	if starttime is None:	×
UNCOV 455	starttime = datetime.now().strftime('%Y/%m/%d %H')	×
UNCOV 456	args.append('starttime="{0}"'.format(starttime))	×
457
UNCOV 458	return common.response_setter(self._run(args))	×
459
460	def has_foreign_configurations(self, securitykey: Optional[str] = None) -> bool:	4✔
461	"""(bool): true if controller has foreign configurations
462	"""
463	args = [	4✔
464	'/fall',
465	'show'
466	]
467
468	if securitykey:	4✔
UNCOV 469	args.append(f'securitykey={securitykey}')	×
470
471	try:	4✔
472	fc_data = common.response_data(self._run(args))	4✔
473	fcs = 0	4✔
474
475	if 'Total foreign Drive Groups' in fc_data:	4✔
476	fcs = int(fc_data['Total foreign Drive Groups'])	4✔
477	if 'Total Foreign PDs' in fc_data:	4✔
478	fcs += int(fc_data['Total Foreign PDs'])	4✔
479	if 'Total Locked Foreign PDs' in fc_data:	4✔
480	fcs += int(fc_data['Total Locked Foreign PDs'])	4✔
481
482	if fcs > 0:	4✔
483	return True	4✔
UNCOV 484	except KeyError:	×
UNCOV 485	pass	×
486	return False	×
487
488	def is_foreign_configuration_healthy(self, securitykey: Optional[str] = None) -> bool:	4✔
489	"""(bool): true if controller has healthy foreign configurations
490	"""
491
492	if not self.has_foreign_configurations(securitykey):	4✔
UNCOV 493	return True	×
494
495	args = [	4✔
496	'/fall',
497	'show'
498	]
499
500	if securitykey:	4✔
UNCOV 501	args.append(f'securitykey={securitykey}')	×
502
503	try:	4✔
504	fc_data = common.response_data(	4✔
505	self._run(args, allow_error_codes=[]))
506	except exc.StorCliCmdErrorCode as e:	4✔
507	if e.error_code == StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION:	4✔
508	return False	4✔
509
UNCOV 510	raise e	×
511
512	return True	4✔
513
514	def delete_foreign_configurations(self, securitykey: Optional[str] = None):	4✔
515	"""Deletes foreign configurations
516
517	Returns:
518	(dict): response cmd data
519	"""
520	args = [	×
521	'/fall',
522	'del'
523	]
524
UNCOV 525	if securitykey:	×
UNCOV 526	args.append(f'securitykey={securitykey}')	×
UNCOV 527	return common.response_cmd(self._run(args))	×
528
529	def import_foreign_configurations(self, securitykey: Optional[str] = None):	4✔
530	"""Imports foreign configurations
531
532	Returns:
533	(dict): response cmd data
534	"""
UNCOV 535	args = [	×
536	'/fall',
537	'import'
538	]
539	if securitykey:	×
540	args.append(f'securitykey={securitykey}')	×
541	return common.response_cmd(self._run(args))	×
542
543	def set_controller_mode(self, mode):	4✔
544	"""Set controller mode command
545
546	Returns:
547	(dict): response cmd data
548	"""
NEW 549	args = [	×
550	'set',
551	f'personality={mode}'
552	]
NEW 553	return common.response_cmd(self._run(args))	×
554
555	def delete_vds(self):	4✔
556	"""Delete all virtual drives
557
558	Returns:
559	(dict): response cmd data
560	"""
NEW 561	args = [	×
562	'/vall',
563	'delete'
564	]
NEW 565	return common.response_cmd(self._run(args))	×
566
567	@property	4✔
568	def phyerrorcounters(self):	4✔
569	"""Get Phy Error Counters
570
571	Returns:
572	(dict): response cmd data
573	"""
NEW 574	args = [	×
575	'/pall',
576	'show'
577	]
NEW UNCOV 578	try:	×
579	# RAID
NEW UNCOV 580	return common.response_data(self._run(args + ['all']))['Phy Error Counters']	×
NEW UNCOV 581	except exc.StorCliCmdError:	×
582	# HBA
NEW UNCOV 583	return common.response_data(self._run(args + ['phyerrorcounters']))	×
584
585
586	class Controllers(object):	4✔
587	"""StorCLI Controllers	4✔
588
589	Instance of this class is iterable with :obj:Controller as item
590
591	Args:
592	binary (str): storcli binary or full path to the binary
593
594	Properties:
595	ids (list of str): list of controllers id
596
597	Methods:
598	get_clt (:obj:Controller): return controller object by id
599	"""
600
601	def __init__(self, binary='storcli64'):	4✔
602	"""Constructor - create StorCLI Controllers object
603
604	Args:
605	binary (str): storcli binary or full path to the binary
606	"""
607	self._binary = binary	4✔
608	self._storcli = StorCLI(binary)	4✔
609
610	@ property	4✔
611	def show(self) -> List[dict]:	4✔
612	"""Static show output to allow getting info of static attributes
613	without the need to touch cached responses
614	Return:
615	list: list of dicts of controllers data from show command
616	"""
617	if not getattr(self, '_show', None):	4✔
618	out = self._storcli.run(['show'], allow_error_codes=[	4✔
619	StorcliErrorCode.INCOMPLETE_FOREIGN_CONFIGURATION])
620	response = common.response_data(out)	4✔
621	if "Number of Controllers" in response and response["Number of Controllers"] == 0:	4✔
622	self._show = []	4✔
623	else:
624	self._show = common.response_data_subkey(	4✔
625	out, ['System Overview', 'IT System Overview'])
626	return self._show	4✔
627
628	@ property	4✔
629	def _ctl_ids(self) -> List[int]:	4✔
630	out = self.show	4✔
631	return [] if not out else [ctl['Ctl'] for ctl in out]	4✔
632
633	@ property	4✔
634	def _ctls(self):	4✔
635	for ctl_id in self._ctl_ids:	4✔
636	yield Controller(ctl_id=ctl_id, binary=self._binary)	4✔
637
638	def __iter__(self):	4✔
639	return self._ctls	4✔
640
641	@ property	4✔
642	def ids(self):	4✔
643	"""(list of str): controllers id
644	"""
645	return self._ctl_ids	4✔
646
647	def get_ctl(self, ctl_id: int) -> Optional[Controller]:	4✔
648	"""Get controller object by id
649
650	Args:
651	ctl_id (str): controller id
652
653	Returns:
654	(None): no controller with id
655	(:obj:Controller): controller object
656	"""
657	for ctl in self:	4✔
658	if ctl.id == ctl_id:	4✔
659	return ctl	4✔
UNCOV 660	return None	×

Naudit / pystorcli2 / 5962669522

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous