Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use iscsi_logout method in autotest

Best Python code snippet using autotest_python

iscsi.py

Source:iscsi.py

...44    target_login = ""45    if "successful" in output:46        target_login = target_name47    return target_login48def iscsi_logout(target_name=None):49    """50    Logout from a target. If the target name is not set then logout all51    targets.52    :params target_name: Name of the target.53    """54    if target_name:55        cmd = "iscsiadm --mode node --logout -T %s" % target_name56    else:57        cmd = "iscsiadm --mode node --logout all"58    output = utils.system_output(cmd)59    target_logout = ""60    if "successful" in output:61        target_logout = target_name62    return target_logout63def iscsi_discover(portal_ip):64    """65    Query from iscsi server for available targets66    :param portal_ip: Ip for iscsi server67    """68    cmd = "iscsiadm -m discovery -t sendtargets -p %s" % portal_ip69    output = utils.system_output(cmd, ignore_status=True)70    session = ""71    if "Invalid" in output:72        logging.debug(output)73    else:74        session = output75    return session76class Iscsi(object):77    """78    Basic iscsi support class. Will handle the emulated iscsi export and79    access to both real iscsi and emulated iscsi device.80    """81    def __init__(self, params, root_dir="/tmp"):82        os_dep.command("iscsiadm")83        self.target = params.get("target")84        self.export_flag = False85        if params.get("portal_ip"):86            self.portal_ip = params.get("portal_ip")87        else:88            self.portal_ip = utils.system_output("hostname")89        if params.get("iscsi_thread_id"):90            self.id = params.get("iscsi_thread_id")91        else:92            self.id = utils.generate_random_string(4)93        self.initiator = params.get("initiator")94        if params.get("emulated_image"):95            self.initiator = None96            os_dep.command("tgtadm")97            emulated_image = params.get("emulated_image")98            self.emulated_image = os.path.join(root_dir, emulated_image)99            self.emulated_id = ""100            self.emulated_size = params.get("image_size")101            self.unit = self.emulated_size[-1].upper()102            self.emulated_size = self.emulated_size[:-1]103            # maps K,M,G,T => (count, bs)104            emulated_size = {'K': (1, 1),105                             'M': (1, 1024),106                             'G': (1024, 1024),107                             'T': (1024, 1048576),108                             }109            if self.unit in emulated_size:110                block_size = emulated_size[self.unit][1]111                size = int(self.emulated_size) * emulated_size[self.unit][0]112                self.create_cmd = ("dd if=/dev/zero of=%s count=%s bs=%sK"113                                   % (self.emulated_image, size, block_size))114    def logged_in(self):115        """116        Check if the session is login or not.117        """118        sessions = iscsi_get_sessions()119        login = False120        if self.target in map(lambda x: x[1], sessions):121            login = True122        return login123    def portal_visible(self):124        """125        Check if the portal can be found or not.126        """127        return bool(re.findall("%s$" % self.target,128                               iscsi_discover(self.portal_ip), re.M))129    def login(self):130        """131        Login session for both real iscsi device and emulated iscsi. Include132        env check and setup.133        """134        login_flag = False135        if self.portal_visible():136            login_flag = True137        elif self.initiator:138            logging.debug("Try to update iscsi initiatorname")139            cmd = "mv /etc/iscsi/initiatorname.iscsi "140            cmd += "/etc/iscsi/initiatorname.iscsi-%s" % self.id141            utils.system(cmd)142            fd = open("/etc/iscsi/initiatorname.iscsi", 'w')143            fd.write("InitiatorName=%s" % self.initiator)144            fd.close()145            utils.system("service iscsid restart")146            if self.portal_visible():147                login_flag = True148        elif self.emulated_image:149            self.export_target()150            utils.system("service iscsid restart")151            if self.portal_visible():152                login_flag = True153        if login_flag:154            iscsi_login(self.target)155    def get_device_name(self):156        """157        Get device name from the target name.158        """159        cmd = "iscsiadm -m session -P 3"160        device_name = ""161        if self.logged_in():162            output = utils.system_output(cmd)163            pattern = r"Target:\s+%s.*?disk\s(\w+)\s+\S+\srunning" % self.target164            device_name = re.findall(pattern, output, re.S)165            try:166                device_name = "/dev/%s" % device_name[0]167            except IndexError:168                logging.error("Can not find target '%s' after login.", self.target)169        else:170            logging.error("Session is not logged in yet.")171        return device_name172    def get_target_id(self):173        """174        Get target id from image name. Only works for emulated iscsi device175        """176        cmd = "tgtadm --lld iscsi --mode target --op show"177        target_info = utils.system_output(cmd)178        target_id = ""179        for line in re.split("\n", target_info):180            if re.findall("Target\s+(\d+)", line):181                target_id = re.findall("Target\s+(\d+)", line)[0]182            if re.findall("Backing store path:\s+(/+.+)", line):183                if self.emulated_image in line:184                    break185        else:186            target_id = ""187        return target_id188    def export_target(self):189        """190        Export target in localhost for emulated iscsi191        """192        if not os.path.isfile(self.emulated_image):193            utils.system(self.create_cmd)194        cmd = "tgtadm --lld iscsi --mode target --op show"195        try:196            output = utils.system_output(cmd)197        except error.CmdError:198            utils.system("service tgtd restart")199            output = utils.system_output(cmd)200        if not re.findall("%s$" % self.target, output, re.M):201            logging.debug("Need to export target in host")202            output = utils.system_output(cmd)203            used_id = re.findall("Target\s+(\d+)", output)204            emulated_id = 1205            while str(emulated_id) in used_id:206                emulated_id += 1207            self.emulated_id = str(emulated_id)208            cmd = "tgtadm --mode target --op new --tid %s" % self.emulated_id209            cmd += " --lld iscsi --targetname %s" % self.target210            utils.system(cmd)211            cmd = "tgtadm --lld iscsi --op bind --mode target "212            cmd += "--tid %s -I ALL" % self.emulated_id213            utils.system(cmd)214        else:215            target_strs = re.findall("Target\s+(\d+):\s+%s$" %216                                     self.target, output, re.M)217            self.emulated_id = target_strs[0].split(':')[0].split()[-1]218        cmd = "tgtadm --lld iscsi --mode target --op show"219        try:220            output = utils.system_output(cmd)221        except error.CmdError:   # In case service stopped222            utils.system("service tgtd restart")223            output = utils.system_output(cmd)224        # Create a LUN with emulated image225        if re.findall(self.emulated_image, output, re.M):226            # Exist already227            logging.debug("Exported image already exists.")228            self.export_flag = True229            return230        else:231            luns = len(re.findall("\s+LUN:\s(\d+)", output, re.M))232            cmd = "tgtadm --mode logicalunit --op new "233            cmd += "--tid %s --lld iscsi " % self.emulated_id234            cmd += "--lun %s " % luns235            cmd += "--backing-store %s" % self.emulated_image236            utils.system(cmd)237            self.export_flag = True238    def delete_target(self):239        """240        Delete target from host.241        """242        cmd = "tgtadm --lld iscsi --mode target --op show"243        output = utils.system_output(cmd)244        if re.findall("%s$" % self.target, output, re.M):245            if self.emulated_id:246                cmd = "tgtadm --lld iscsi --mode target --op delete "247                cmd += "--tid %s" % self.emulated_id248                utils.system(cmd)249    def logout(self):250        """251        Logout from target.252        """253        if self.logged_in():254            iscsi_logout(self.target)255    def cleanup(self):256        """257        Clean up env after iscsi used.258        """259        self.logout()260        if os.path.isfile("/etc/iscsi/initiatorname.iscsi-%s" % self.id):261            cmd = " mv /etc/iscsi/initiatorname.iscsi-%s" % self.id262            cmd += " /etc/iscsi/initiatorname.iscsi"263            utils.system(cmd)264            cmd = "service iscsid restart"265            utils.system(cmd)266        if self.export_flag:...

tests.py

Source:tests.py

...62    def setUp(self):63        if Global.debug or Global.verbosity > 1:64            # this makes debug printing a little more clean65            print('', file=sys.stderr)66    def iscsi_logout(self):67        res = util.run_cmd(['iscsiadm', '-m', 'node',68                            '-T', Global.target,69                            '-p', Global.ipnr,70                            '--logout'])71        if res not in [0, 21]:72            self.fail('logout failed')73        self.assertFalse(os.path.exists(Global.device), '%s: exists after logout!' % Global.device)74    def test_InitialR2T_on_ImmediateData_off(self):75        """Test Initial Request to Transmit on, but Immediate Data off"""76        i = 177        for v in self.param_values:78            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):79                if i not in Global.subtest_list:80                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)81                else:82                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)83                    self.iscsi_logout()84                    iscsi_data = IscsiData('No', 'Yes', 'None', 'None', v[0], v[1], v[2])85                    iscsi_data.update_cfg(Global.target, Global.ipnr)86                    self.run_the_rest()87            i += 188    def test_InitialR2T_off_ImmediateData_on(self):89        """Test Initial Request to Transmit off, Immediate Data on"""90        i = 191        for v in self.param_values:92            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):93                if i not in Global.subtest_list:94                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)95                else:96                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)97                    self.iscsi_logout()98                    iscsi_data = IscsiData('Yes', 'No', 'None', 'None', v[0], v[1], v[2])99                    iscsi_data.update_cfg(Global.target, Global.ipnr)100                    self.run_the_rest()101            i += 1102    def test_InitialR2T_on_ImmediateData_on(self):103        """Test Initial Request to Transmit and Immediate Data on"""104        i = 1105        for v in self.param_values:106            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):107                if i not in Global.subtest_list:108                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)109                else:110                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)111                    self.iscsi_logout()112                    iscsi_data = IscsiData('Yes', 'Yes', 'None', 'None', v[0], v[1], v[2])113                    iscsi_data.update_cfg(Global.target, Global.ipnr)114                    self.run_the_rest()115            i += 1116    def test_InitialR2T_off_ImmediateData_off(self):117        """Test Initial Request to Transmit and Immediate Data off"""118        i = 1119        for v in self.param_values:120            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):121                if i not in Global.subtest_list:122                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)123                else:124                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)125                    self.iscsi_logout()126                    iscsi_data = IscsiData('No', 'No', 'None', 'None', v[0], v[1], v[2])127                    iscsi_data.update_cfg(Global.target, Global.ipnr)128                    self.run_the_rest()129            i += 1130    def test_HdrDigest_on_DataDigest_off(self):131        """Test With Header Digest"""132        i = 1133        for v in self.param_values:134            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):135                if i not in Global.subtest_list:136                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)137                else:138                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)139                    self.iscsi_logout()140                    iscsi_data = IscsiData('No', 'Yes', 'CRC32C', 'None', v[0], v[1], v[2])141                    iscsi_data.update_cfg(Global.target, Global.ipnr)142                    self.run_the_rest()143            i += 1144    def test_HdrDigest_on_DataDigest_on(self):145        """Test With Header Digest"""146        i = 1147        for v in self.param_values:148            with self.subTest('Testing FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v), i=i):149                if i not in Global.subtest_list:150                    util.vprint('Skipping subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)151                else:152                    util.vprint('Running subtest %d: FirstBurst={} MaxBurst={} MaxRecv={}'.format(*v) % i)153                    self.iscsi_logout()154                    iscsi_data = IscsiData('No', 'Yes', 'CRC32C', 'CRC32C', v[0], v[1], v[2])155                    iscsi_data.update_cfg(Global.target, Global.ipnr)156                    self.run_the_rest()157            i += 1158    def run_the_rest(self):159        res = util.run_cmd(['iscsiadm', '-m', 'node',160                            '-T', Global.target,161                            '-p', Global.ipnr,162                            '--login'])163        self.assertEqual(res, 0, 'cannot login to device')164        # wait a few seconds for the device to show up165        if not util.wait_for_path(Global.device):166            self.fail('%s: does not exist after login' % Global.device)167        # run parted to partition the disc with one whole disk partition...

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.