12162379188

Committed 28 May 2024 06:08PM UTC coverage: 61.359% (+0.02%) from 61.34%

Build # 12162379188

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #19 from liveaction/develop

feat: add support to configure controller jbod mode

Run Details

5 of 12 new or added lines in 1 file covered. (41.67%)

2 existing lines in 1 file now uncovered.

1129 of 1840 relevant lines covered (61.36%)

2.39 hits per line

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

59.31

/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
        jbod (str): enables/disables JBOD mode; by default, drives become system drives.

    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 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))

    @property
    @common.lower
    def jbod(self):
        """Get/Set jbod mode state

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

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

        for pr in common.response_property(self._run(args)):
            if pr['Ctrl_Prop'] == "JBOD":
                return pr['Value']
        return 'off'

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


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 _ctl_ids(self) -> List[int]:
        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:
            return []
        else:
            return [ctl['Ctl'] for ctl in common.response_data_subkey(out, ['System Overview', 'IT System Overview'])]

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

Naudit / pystorcli2 / 12162379188

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