Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

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.