Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use _runcmd method in fMBT
Best Python code snippet using fMBT_python
init.py
Source:init.py 
1# Copyright 2014 Google Inc. All Rights Reserved.2"""Workflow to set up gcloud environment."""3import argparse4import os5import sys6import types7from googlecloudsdk.calliope import base8from googlecloudsdk.calliope import exceptions as c_exc9from googlecloudsdk.core import log10from googlecloudsdk.core import properties11from googlecloudsdk.core.console import console_io12from googlecloudsdk.core.util import files13class Init(base.Command):14 """Initialize or reinitialize gcloud."""15 detailed_help = {16 'DESCRIPTION': """\17 {description}18 {command} launches a interactive getting-started gcloud19 workflow, and replaces `gcloud auth login` as the recommended20 command to execute after newly installing gcloud. This workflow21 performs a variety of setup tasks, including the following:22 - launching an authorization flow or selecting credentials23 - setting properties including project, default GCE zone, and24 default GCE region25 - suggesting cloning a source repository26 Most users will run {command} to get started with gcloud. Subsequent27 {command} invocations can be use to create new gcloud configurations28 or to reinitialize existing configurations. See `gcloud topic29 configurations` for additional information about configurations.30 Properties set by `gcloud init` are local and persistent. They are31 not affected by remote changes to your project. For instance, your32 configuration's default Compute Engine zone will remain stable, even33 if you or another user changes the project default zone in the34 Developer Console website. You can resync your configuration at any35 time by rerunning `gcloud init`.36 """,37 }38 @staticmethod39 def Args(parser):40 parser.add_argument(41 'obsolete_project_arg',42 nargs='?',43 help=argparse.SUPPRESS)44 parser.add_argument(45 '--console-only',46 action='store_true',47 help=('Don\'t launch a browser for authentication.'))48 def Run(self, args):49 """Allows user to select configuration, and initialize it."""50 if args.obsolete_project_arg:51 raise c_exc.InvalidArgumentException(52 args.obsolete_project_arg,53 '`gcloud init` has changed and no longer takes a PROJECT argument. '54 'Please use `gcloud source repos clone` to clone this '55 'project\'s source repositories.')56 log.status.write('Welcome! This command will take you through '57 'the configuration of gcloud.\n\n')58 if properties.VALUES.core.disable_prompts.GetBool():59 raise c_exc.InvalidArgumentException(60 'disable_prompts/--quiet',61 'gcloud init command cannot run with disabled prompts.')62 configuration_name = None63 try:64 configuration_name = self._PickConfiguration()65 if not configuration_name:66 return67 log.status.write('Your current configuration has been set to: [{0}]\n\n'68 .format(configuration_name))69 if not self._PickAccount(args.console_only):70 return71 if not self._PickProject():72 return73 self._PickDefaultRegionAndZone()74 self._PickRepo()75 log.status.write('\ngcloud has now been configured!\n')76 finally:77 log.status.write('You can use [gcloud config] to '78 'change more gcloud settings.\n\n')79 if configuration_name:80 log.status.write('Your current configuration is: [{0}]\n\n'81 .format(configuration_name))82 log.status.flush()83 # Not using self._RunCmd to get command actual output.84 self.cli.Execute(['config', 'list'])85 def _PickAccount(self, console_only):86 """Checks if current credentials are valid, if not runs auth login.87 Args:88 console_only: bool, True if the auth flow shouldn't use the browser89 Returns:90 bool, True if valid credentials are setup.91 """92 auth_info = self._RunCmd(['auth', 'list'])93 if auth_info and auth_info.accounts:94 idx = console_io.PromptChoice(95 auth_info.accounts + ['Login with new credentials'],96 message='Pick credentials to use:',97 prompt_string=None)98 if idx is None:99 return None100 new_credentials = idx == len(auth_info.accounts)101 else:102 answer = console_io.PromptContinue(103 prompt_string='To continue, you must login. Would you like to login')104 if not answer:105 return False106 new_credentials = True107 if new_credentials:108 # gcloud auth login may have user interaction, do not suppress it.109 browser_args = ['--no-launch-browser'] if console_only else []110 if not self._RunCmd(['auth', 'login'],111 ['--force', '--brief'] + browser_args,112 disable_user_output=False):113 return None114 else:115 account = auth_info.accounts[idx]116 self._RunCmd(['config', 'set'], ['account', account])117 log.status.write('You are now logged in as: [{0}]\n\n'118 .format(properties.VALUES.core.account.Get()))119 return True120 def _PickConfiguration(self):121 """Allows user to re-initialize, create or pick new configuration.122 Returns:123 Configuration name or None.124 """125 configs = self._RunCmd(['config', 'configurations', 'list'])126 if not configs:127 new_config_name = 'default'128 if self._RunCmd(['config', 'configurations', 'create'],129 [new_config_name]):130 self._RunCmd(['config', 'configurations', 'activate'],131 [new_config_name])132 properties.PropertiesFile.Invalidate()133 return new_config_name134 config_names = [cfg.name for cfg in configs]135 active_configs = [cfg.name for cfg in configs136 if getattr(cfg, 'is_active', False)]137 if not active_configs:138 return None139 choices = []140 active_config = active_configs[0]141 log.status.write('Settings from your current configuration [{0}] are:\n'142 .format(active_config))143 log.status.flush()144 # Not using self._RunCmd to get command actual output.145 self.cli.Execute(['config', 'list'])146 log.out.flush()147 log.status.write('\n')148 log.status.flush()149 choices.append(150 'Re-initialize this configuration [{0}] with new settings '.format(151 active_config))152 choices.append('Create a new configuration')153 config_choices = [name for name in config_names if name != active_config]154 choices.extend('Switch to and re-initialize '155 'existing configuration: [{0}]'.format(name)156 for name in config_choices)157 idx = console_io.PromptChoice(choices, message='Pick configuration to use:')158 if idx is None:159 return None160 if idx == 0: # If reinitialize was selected.161 self._CleanCurrentConfiguration()162 return active_config163 if idx == 1: # Second option is to create new configuration.164 return self._CreateConfiguration()165 config_name = config_choices[idx - 2]166 self._RunCmd(['config', 'configurations', 'activate'], [config_name])167 return config_name168 def _PickProject(self):169 """Allows user to select a project.170 Returns:171 str, project_id or None if was not selected.172 """173 projects = self._RunExperimentalCmd(['beta', 'projects', 'list'])174 if projects is None: # Failed to get the list.175 project_id = console_io.PromptResponse(176 'Enter project id you would like to use: ')177 if not project_id:178 return None179 else:180 projects = sorted(projects, key=lambda prj: prj.projectId)181 choices = ['[{0}]'.format(project.projectId) for project in projects]182 if not choices:183 log.status.write('\nThis account has no projects. Please create one in '184 'developers console '185 '(https://console.developers.google.com/project) '186 'before running this command.\n')187 return None188 if len(choices) == 1:189 project_id = projects[0].projectId190 else:191 idx = console_io.PromptChoice(192 choices,193 message='Pick cloud project to use: ',194 prompt_string=None)195 if idx is None:196 return197 project_id = projects[idx].projectId198 self._RunCmd(['config', 'set'], ['project', project_id])199 log.status.write('Your current project has been set to: [{0}].\n\n'200 .format(project_id))201 return project_id202 def _PickDefaultRegionAndZone(self):203 """Pulls metadata properties for region and zone and sets them in gcloud."""204 try:205 project_info = self._RunCmd(['compute', 'project-info', 'describe'])206 except c_exc.FailedSubCommand:207 log.status.write('Not setting default zone/region.\nMake sure Compute '208 'Engine API is enabled for your project.\n\n')209 return None210 default_zone = None211 default_region = None212 if project_info is not None:213 metadata = project_info.get('commonInstanceMetadata', {})214 for item in metadata.get('items', []):215 if item['key'] == 'google-compute-default-zone':216 default_zone = item['value']217 elif item['key'] == 'google-compute-default-region':218 default_region = item['value']219 # Same logic applies to region and zone properties.220 def SetProperty(name, default_value, list_command):221 """Set named compute property to default_value or get via list command."""222 if default_value:223 log.status.write('Your project default compute {0} has been set to '224 '[{1}].\nYou can change it by running '225 '[gcloud config set compute/{0} NAME].\n\n'226 .format(name, default_value['name']))227 else:228 values = self._RunCmd(list_command)229 if values is None:230 return231 values = list(values)232 idx = console_io.PromptChoice(233 ['[{0}]'.format(value['name']) for value in values]234 + ['Do not set default {0}'.format(name)],235 message=('Which compute {0} would you like '236 'to use as project default?'.format(name)),237 prompt_string=None)238 if idx is None or idx == len(values):239 return240 default_value = values[idx]241 self._RunCmd(['config', 'set'],242 ['compute/{0}'.format(name), default_value['name']])243 return default_value244 if default_zone:245 default_zone = self._RunCmd(['compute', 'zones', 'describe'],246 [default_zone])247 zone = SetProperty('zone', default_zone, ['compute', 'zones', 'list'])248 if zone and not default_region:249 default_region = zone['region']250 if default_region:251 default_region = self._RunCmd(['compute', 'regions', 'describe'],252 [default_region])253 SetProperty('region', default_region, ['compute', 'regions', 'list'])254 def _PickRepo(self):255 """Allows user to clone one of the projects repositories."""256 cmd = ['alpha', 'source', 'repos', 'list']257 repos = self._RunExperimentalCmd(cmd)258 if repos:259 repos = sorted(repo.name or 'default' for repo in repos)260 log.status.write(261 'This project has one or more associated git repositories.\n')262 idx = console_io.PromptChoice(263 ['[{0}]'.format(repo) for repo in repos] + ['Do not clone'],264 message='Pick repository to clone to your local machine:',265 prompt_string=None)266 if idx >= 0 and idx < len(repos):267 repo_name = repos[idx]268 else:269 return270 elif repos is None:271 log.status.write('Could not retrieve list of repos via [gcloud {0}]\n'272 .format(' '.join(cmd)))273 log.status.write('Perhaps alpha commands are not enabled '274 'or the repos list command failed.\n'275 '\n')276 answer = console_io.PromptContinue(277 prompt_string='Generally projects have a repository named [default]. '278 'Would you like to try clone it?')279 if not answer:280 return281 repo_name = 'default'282 else:283 return284 self._CloneRepo(repo_name)285 def _CloneRepo(self, repo_name):286 """Queries user for output path and clones selected repo to it."""287 default_clone_path = os.path.join(os.getcwd(), repo_name)288 while True:289 clone_path = console_io.PromptResponse(290 'Where would you like to clone [{0}] repository to [{1}]:'291 .format(repo_name, default_clone_path))292 if not clone_path:293 clone_path = default_clone_path294 if os.path.exists(clone_path):295 log.status.write('Directory [{0}] already exists\n'.format(clone_path))296 continue297 parent_dir = os.path.dirname(clone_path)298 if not os.path.isdir(parent_dir):299 log.status.write('No such directory [{0}]\n'.format(parent_dir))300 answer = console_io.PromptContinue(301 prompt_string='Would you like to create it')302 if answer:303 files.MakeDir(parent_dir)304 break305 else:306 break307 self._RunCmd(['source', 'repos', 'clone'], [repo_name, clone_path])308 log.status.write('\nGit repository has been cloned to [{0}]\n'309 .format(clone_path))310 def _CreateConfiguration(self):311 configuration_name = console_io.PromptResponse(312 'Enter configuration name: ')313 new_config_name = self._RunCmd(['config', 'configurations', 'create'],314 [configuration_name])315 if new_config_name:316 self._RunCmd(['config', 'configurations', 'activate'],317 [configuration_name])318 properties.PropertiesFile.Invalidate()319 return new_config_name320 def _CleanCurrentConfiguration(self):321 self._RunCmd(['config', 'unset'], ['account'])322 self._RunCmd(['config', 'unset'], ['project'])323 self._RunCmd(['config', 'unset'], ['compute/zone'])324 self._RunCmd(['config', 'unset'], ['compute/region'])325 def _RunCmd(self, cmd, params=None, disable_user_output=True):326 if not self.cli.IsValidCommand(cmd):327 log.info('Command %s does not exist.', cmd)328 return None329 if params is None:330 params = []331 args = cmd + params332 log.info('Executing: [gcloud %s]', ' '.join(args))333 try:334 # Disable output from individual commands, so that we get335 # command run results, and don't clutter output of init.336 if disable_user_output:337 args.append('--no-user-output-enabled')338 if (properties.VALUES.core.verbosity.Get() is None and339 disable_user_output):340 # Unless user explicitly set verbosity, suppress from subcommands.341 args.append('--verbosity=none')342 result = self.cli.Execute(args)343 # Best effort to force result of Execute eagerly. Don't just check344 # that result is iterable to avoid category errors (e.g., accidently345 # converting a string or dict to a list).346 if type(result) is types.GeneratorType:347 return list(result)348 return result349 except SystemExit as exc:350 log.status.write('[{0}] has failed\n'.format(' '.join(cmd + params)))351 raise c_exc.FailedSubCommand(cmd + params, exc.code)352 except BaseException:353 log.status.write('Failed to run [{0}]\n'.format(' '.join(cmd + params)))354 raise355 def _RunExperimentalCmd(self, cmd, params=None):356 try:357 return self._RunCmd(cmd, params)358 except (Exception) as e: # pylint:disable=broad-except359 cmd_string = ' '.join(cmd + (params or []))360 log.status.write(361 'Unexpected failure while executing [{0}]: [{1}: {2}]\n'362 'Please report by running `gcloud feedback`.\n\n'.format(363 cmd_string, type(e), e))364 log.debug('Failed to execute %s, %s, %s, %s', cmd_string, *sys.exc_info())...
hostfirewall.py
Source:hostfirewall.py
...10# Rule syntax: 11# 'all'12# proto[:port], e.g., tcp:80, icmp:*, udp:*, icmp, udp13#14def _runcmd(progargs, stdinput=None):15 '''16 Run the command progargs with optional input to be fed in to stdin.17 '''18 stdin = None19 if stdinput is not None:20 assert(isinstance(stdinput, list))21 stdin=PIPE22 err = 023 output = b''24 log_debug("Calling {} with input {}".format(' '.join(progargs), stdinput))25 try:26 p = Popen(progargs, shell=True, stdin=stdin, 27 stderr=STDOUT, stdout=PIPE, universal_newlines=True)28 if stdinput is not None:29 for cmd in stdinput:30 print(cmd, file=p.stdin)31 p.stdin.close()32 output = p.stdout.read()33 p.stdout.close()34 err = p.wait(timeout=1.0)35 except OSError as e:36 err = e.errno37 log_warn("Error calling {}: {}".format(progargs, e.stderror))38 except Exception as e:39 errstr = str(e)40 log_warn("Error calling {}: {}".format(progargs, errstr))41 err = -142 log_debug("Result of command (errcode {}): {}".format(err, output))43 return err,output44class Firewall(object):45 _instance = None46 def __init__(self, interfaces, rules):47 if Firewall._instance:48 raise Exception("Firewall can only be instantiated once.")49 Firewall._instance = self50 cls = _osmap.get(sys.platform, None)51 if cls is None:52 raise Exception("{} can't run on {}".format(self.__class__.__name__, sys.platform))53 self._firewall_delegate = cls(interfaces, rules)54 def __enter__(self):55 self._firewall_delegate.block()56 if VerboseOutput.enabled():57 self._firewall_delegate.show_rules()58 return None59 def __exit__(self, exctype, excvalue, traceback):60 self._firewall_delegate.unblock()61 return None62 @staticmethod63 def add_rule(rule):64 Firewall._instance._firewall_delegate.add_rule(rule)65class AbstractFirewall(metaclass=ABCMeta):66 def __init__(self, interfaces, rules):67 self._rules = []68 @staticmethod69 def _interp_rule(rule):70 mobj = re.match('(?P<proto>tcp|udp|icmp)(:(?P<port>\d+|\*))?', rule)71 if mobj is None:72 raise ValueError("Can't parse rule: {}".format(rule))73 d = mobj.groupdict()74 proto = d['proto']75 port = d['port']76 if port == '*':77 port = None78 return proto,port79 @abstractmethod80 def block(self):81 pass82 @abstractmethod83 def unblock(self):84 pass85 @abstractmethod86 def add_rule(self, rule):87 pass88 @abstractmethod89 def show_rules(self):90 pass91class TestModeFirewall(AbstractFirewall):92 def __init__(self, interfaces, rules):93 super().__init__(interfaces, rules)94 for r in rules:95 self.add_rule(r)96 def block(self):97 pass98 def unblock(self):99 pass100 def show_rules(self):101 pass102 def add_rule(self, rule):103 if rule.strip() == 'all':104 proto,port = 'all',None105 elif rule.strip() == 'none':106 proto,port = 'none',None107 else:108 proto,port = self._interp_rule(rule)109 self._rules.append((proto,port))110class LinuxFirewall(AbstractFirewall):111 def __init__(self, interfaces, rules):112 super().__init__(interfaces, rules)113 self._intf = deepcopy(list(interfaces))114 st,output = _runcmd("/sbin/iptables-save")115 self._saved_iptables = output116 self._arpignore = {}117 self._rulecmds = [ '/sbin/iptables -F', '/sbin/iptables -t raw -F' ]118 # --protocol {} -i {} --port {}119 doall = False120 for r in rules:121 cmds = self._parse_rule(r)122 self._rulecmds.extend(cmds)123 if r == 'all':124 doall = True125 if doall:126 badintf = []127 for intf in interfaces:128 st,output = _runcmd('/sbin/sysctl net.ipv4.conf.{}.arp_ignore'.format(intf))129 if st != 0:130 badintf.append(intf)131 continue132 self._arpignore[intf] = int(output.split()[-1])133 st,output = _runcmd('/sbin/sysctl -w net.ipv4.conf.{}.arp_ignore=8'.format(intf))134 for intf in badintf:135 self._intf.remove(intf) # alias of interfaces, so just remove136 # from self._intf137 log_debug("Commands for firewall: {}".format(self._rulecmds))138 def _parse_rule(self, rule):139 cmds = []140 if rule.strip() == 'none':141 pass142 elif rule.strip() == 'all':143 for intf in self._intf:144 cmds.append('/sbin/iptables -t raw -A PREROUTING -j DROP -i {}'.format(intf)) 145 else: 146 proto,port = self._interp_rule(rule)147 if port is not None:148 portpart = " --dport {}".format(port)149 else:150 portpart = "" 151 for intf in self._intf:152 cmds.append('/sbin/iptables -t raw -A PREROUTING -j DROP --protocol {} -i {}{}'.format(153 proto, intf, portpart))154 return cmds155 def add_rule(self, rule):156 for cmd in self._parse_rule(rule):157 st,output = _runcmd(cmd)158 self._rulecmds.append(cmd)159 log_debug("Adding firewall rule: {}".format(cmd))160 def block(self):161 log_info("Saving iptables state and installing switchyard rules")162 for cmd in self._rulecmds:163 st,output = _runcmd(cmd)164 def unblock(self):165 # clear switchyard tables, load up saved state166 log_info("Restoring saved iptables state")167 st,output = _runcmd("/sbin/iptables -F")168 st,output = _runcmd("/sbin/iptables -t raw -F")169 st,output = _runcmd("/sbin/iptables-restore", [self._saved_iptables])170 for intf in self._intf:171 if intf in self._arpignore:172 st,output = _runcmd('/sbin/sysctl -w net.ipv4.conf.{}.arp_ignore={}'.format(intf, self._arpignore[intf]))173 def show_rules(self):174 st,output = _runcmd("/sbin/iptables -t raw -n --list")175 output = output.strip()176 log_info("Rules installed: {}".format(output)) 177class MacOSFirewall(AbstractFirewall):178 def __init__(self, interfaces, rules):179 super().__init__(interfaces, rules)180 self._interfaces = interfaces181 for r in rules:182 cmds = self._parse_rule(r)183 self._rules.extend(cmds)184 st,output = _runcmd("/sbin/pfctl -E")185 mobj = re.search("Token\s*:\s*(\d+)", output, re.M)186 if mobj is None:187 raise RuntimeError("Couldn't get pfctl token. Are you running as root?")188 self._token = mobj.groups()[0]189 log_debug("Rules to install: {}".format(self._rules))190 log_info("Enabling pf: {}".format(output.replace('\n', '; ')))191 def _parse_rule(self, rule):192 cmds = []193 if rule.strip() == 'none':194 pass195 elif rule.strip() == 'all':196 rulestr = 'block drop on {} all'197 else:198 proto, port = self._interp_rule(rule)199 if port is not None:200 portpart = " port {}".format(port)201 else:202 portpart = ""203 rulestr = "proto {0} from any{1} to any{1}".format(proto, portpart)204 rulestr = "block drop on {} " + rulestr205 for intf in self._interfaces:206 cmds.append(rulestr.format(intf))207 return cmds208 def add_rule(self, rule):209 cmds = self._parse_rule(rule)210 self._rules.extend(cmds)211 st,output = _runcmd("/sbin/pfctl -aswitchyard -f -", cmds)212 log_debug("Adding firewall rules: {}".format(cmds))213 def block(self):214 '''215 pfctl -a switchyard -f- < rules.txt216 pfctl -a switchyard -F rules217 pfctl -t switchyard -F r218 '''219 st,output = _runcmd("/sbin/pfctl -aswitchyard -f -", self._rules)220 log_debug("Installing rules: {}".format(output))221 def unblock(self):222 '''223 '''224 st,output = _runcmd("/sbin/pfctl -aswitchyard -Fr") # flush rules225 log_debug("Flushing rules: {}".format(output))226 st,output = _runcmd("/sbin/pfctl -X {}".format(self._token))227 log_info("Releasing pf: {}".format(output.replace('\n', '; ')))228 def show_rules(self):229 st,output = _runcmd("/sbin/pfctl -aswitchyard -srules")230 output = output.replace('No ALTQ support in kernel', '')231 output = output.replace('ALTQ related functions disabled', '')232 output = output.strip()233 log_info("Rules installed: {}".format(output)) 234_osmap = {235 'darwin': MacOSFirewall,236 'linux': LinuxFirewall,237 'test': TestModeFirewall,...
iscsiadm.py
Source:iscsiadm.py
1from threading import Lock2import misc3from vdsm import constants4# iscsiadm exit statuses5ISCSI_ERR_SESS_EXISTS = 156ISCSI_ERR_LOGIN_AUTH_FAILED = 247ISCSI_ERR_OBJECT_NOT_FOUND = 218class IscsiError(RuntimeError): pass9class ReservedInterfaceNameError(IscsiError): pass10class IscsiInterfaceError(IscsiError): pass11class IsciInterfaceAlreadyExistsError(IscsiInterfaceError): pass12class IsciInterfaceCreationError(IscsiInterfaceError): pass13class IscsiInterfaceDoesNotExistError(IscsiInterfaceError): pass14class IscsiInterfaceUpdateError(IscsiInterfaceError): pass15class IscsiInterfaceDeletionError(IscsiInterfaceError): pass16class IscsiDiscoverdbError(IscsiError): pass17class IscsiInterfaceListingError(IscsiError): pass18class IscsiAuthenticationError(IscsiError): pass19class IscsiNodeError(IscsiError): pass20class IscsiSessionNotFound(IscsiError): pass21class IscsiSessionError(IscsiError): pass22_RESERVED_INTERFACES = ("default", "tcp", "iser")23# Running multiple iscsiadm commands in parallel causes random problems.24# This serializes all calls to iscsiadm.25# Remove when iscsid is actually thread safe.26_iscsiadmLock = Lock()27def _runCmd(args, hideValue=False):28 # FIXME: I don't use supervdsm because this entire module has to just be29 # run as root and there is no such feature yet in supervdsm. When such30 # feature exists please change this.31 with _iscsiadmLock:32 cmd = [constants.EXT_ISCSIADM] + args33 printCmd = None34 if hideValue:35 printCmd = cmd[:]36 for i, arg in enumerate(printCmd):37 if arg != "-v":38 continue39 if i < (len(printCmd) - 1):40 printCmd[i + 1] = "****"41 return misc.execCmd(cmd, printable=printCmd, sudo=True)42def iface_exists(interfaceName):43 #FIXME: can be optimized by checking /var/lib/iscsi/ifaces44 return interfaceName in iface_list()45def iface_new(name):46 if name in _RESERVED_INTERFACES:47 raise ReservedInterfaceNameError(name)48 rc, out, err = _runCmd(["-m", "iface", "-I", name, "--op=new"])49 if rc == 0:50 return51 if iface_exists(name):52 raise IsciInterfaceAlreadyExistsError(name)53 raise IsciInterfaceCreationError(name, rc, out, err)54def iface_update(name, key, value):55 rc, out, err = _runCmd(["-m", "iface", "-I", name, "-n", key, "-v", value,56 "--op=update"])57 if rc == 0:58 return59 if not iface_exists(name):60 raise IscsiInterfaceDoesNotExistError(name)61 raise IscsiInterfaceUpdateError(name, rc, out, err)62def iface_delete(name):63 rc, out, err = _runCmd(["-m", "iface", "-I", name, "--op=delete"])64 if rc == 0:65 return66 if not iface_exists(name):67 raise IscsiInterfaceDoesNotExistError(name)68 raise IscsiInterfaceDeletionError(name)69def iface_list():70 # FIXME: This can be done more effciently by iterating71 # /var/lib/iscsi/ifaces. Fix if ever a performance bottleneck.72 rc, out, err = _runCmd(["-m", "iface"])73 if rc == 0:74 return [line.split()[0] for line in out]75 raise IscsiInterfaceListingError(rc, out, err)76def iface_info(name):77 # FIXME: This can be done more effciently by reading78 # /var/lib/iscsi/ifaces/<iface name>. Fix if ever a performance bottleneck.79 rc, out, err = _runCmd(["-m", "iface", "-I", name])80 if rc == 0:81 res = {}82 for line in out:83 if line.startswith("#"):84 continue85 key, value = line.split("=", 1)86 res[key.strip()] = value.strip()87 return res88 if not iface_exists(name):89 raise IscsiInterfaceDoesNotExistError(name)90 raise IscsiInterfaceListingError(rc, out, err)91def discoverydb_new(discoveryType, iface, portal):92 rc, out, err = _runCmd(["-m", "discoverydb", "-t", discoveryType, "-I",93 iface, "-p", portal, "--op=new"])94 if rc == 0:95 return96 if not iface_exists(iface):97 raise IscsiInterfaceDoesNotExistError(iface)98 raise IscsiDiscoverdbError(rc, out, err)99def discoverydb_update(discoveryType, iface, portal, key, value, hideValue=False):100 rc, out, err = _runCmd(["-m", "discoverydb", "-t", discoveryType, "-I",101 iface, "-p", portal, "-n", key, "-v", value, "--op=update"], hideValue)102 if rc == 0:103 return104 if not iface_exists(iface):105 raise IscsiInterfaceDoesNotExistError(iface)106 raise IscsiDiscoverdbError(rc, out, err)107def discoverydb_discover(discoveryType, iface, portal):108 rc, out, err = _runCmd(["-m", "discoverydb", "-t", discoveryType, "-I",109 iface, "-p", portal, "--discover"])110 if rc == 0:111 res = []112 for line in out:113 rest, iqn = line.split()114 rest, tpgt = rest.split(",")115 ip, port = rest.split(":")116 res.append((ip, int(port), int(tpgt), iqn))117 return res118 if not iface_exists(iface):119 raise IscsiInterfaceDoesNotExistError(iface)120 if rc == ISCSI_ERR_LOGIN_AUTH_FAILED:121 raise IscsiAuthenticationError(rc, out, err)122 raise IscsiDiscoverdbError(rc, out, err)123def discoverydb_delete(discoveryType, iface, portal):124 rc, out, err = _runCmd(["-m", "discoverydb", "-t", discoveryType, "-I",125 iface, "-p", portal, "--op=delete"])126 if rc == 0:127 return128 if not iface_exists(iface):129 raise IscsiInterfaceDoesNotExistError(iface)130 raise IscsiDiscoverdbError(rc, out, err)131def node_new(iface, portal, targetName):132 rc, out, err = _runCmd(["-m", "node", "-T", targetName, "-I", iface, "-p",133 portal, "--op=new"])134 if rc == 0:135 return136 if not iface_exists(iface):137 raise IscsiInterfaceDoesNotExistError(iface)138 raise IscsiNodeError(rc, out, err)139def node_update(iface, portal, targetName, key, value, hideValue=False):140 rc, out, err = _runCmd(["-m", "node", "-T", targetName, "-I", iface, "-p",141 portal, "-n", key, "-v", value, "--op=update"], hideValue)142 if rc == 0:143 return144 if not iface_exists(iface):145 raise IscsiInterfaceDoesNotExistError(iface)146 raise IscsiNodeError(rc, out, err)147def node_delete(iface, portal, targetName):148 rc, out, err = _runCmd(["-m", "node", "-T", targetName, "-I", iface, "-p",149 portal, "--op=delete"])150 if rc == 0:151 return152 if not iface_exists(iface):153 raise IscsiInterfaceDoesNotExistError(iface)154 raise IscsiNodeError(rc, out, err)155def node_disconnect(iface, portal, targetName):156 rc, out, err = _runCmd(["-m", "node", "-T", targetName, "-I", iface, "-p",157 portal, "-u"])158 if rc == 0:159 return160 if not iface_exists(iface):161 raise IscsiInterfaceDoesNotExistError(iface)162 if rc == ISCSI_ERR_OBJECT_NOT_FOUND:163 raise IscsiSessionNotFound(iface, portal, targetName)164 raise IscsiNodeError(rc, out, err)165def node_login(iface, portal, targetName):166 rc, out, err = _runCmd(["-m", "node", "-T", targetName, "-I", iface, "-p",167 portal, "-l"])168 if rc == 0:169 return170 if not iface_exists(iface):171 raise IscsiInterfaceDoesNotExistError(iface)172 if rc == ISCSI_ERR_LOGIN_AUTH_FAILED:173 raise IscsiAuthenticationError(rc, out, err)174 raise IscsiNodeError(rc, out, err)175def session_rescan():176 rc, out, err = _runCmd(["-m", "session", "-R"])177 if rc == 0:178 return179 raise IscsiSessionError(rc, out, err)180def session_logout(sessionId):181 rc, out, err = _runCmd(["-m", "session", "-r", str(sessionId), "-u"])182 if rc == 0:183 return...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run fMBT automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
