How to use iscsi_logout method in autotest

Best Python code snippet using autotest_python Github


Full Screen

...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 = 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 = params.get("iscsi_thread_id")91 else:92 = 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 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$" %,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( 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.", 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$" %, 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, 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$" %, 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( 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" % 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:...

Full Screen

Full Screen Github


Full Screen

...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',,69 '-p', Global.ipnr,70 '--logout'])71 if res not in [0, 21]:72'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.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.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.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.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.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.ipnr)156 self.run_the_rest()157 i += 1158 def run_the_rest(self):159 res = util.run_cmd(['iscsiadm', '-m', 'node',160 '-T',,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'%s: does not exist after login' % Global.device)167 # run parted to partition the disc with one whole disk partition...

Full Screen

Full Screen

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:


You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run autotest 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!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?