Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use rmdir method in fMBT
Best Python code snippet using fMBT_python
test_rmdir_subvol_down.py
Source:test_rmdir_subvol_down.py 
...91 ret = mkdir(self.clients[0], child_dir)92 self.assertTrue(ret, ('mkdir failed for %s ' % child_dir))93 g.log.info("mkdir of child directory %s successful", child_dir)94 # 'rmdir' on parent should fail with ENOTCONN95 ret = rmdir(self.clients[0], parent_dir)96 self.assertFalse(ret, ('Expected rmdir to fail for %s' % parent_dir))97 g.log.info("rmdir of parent directory %s failed as expected",98 parent_dir)99 # Cleanup100 # Bring up the subvol - restart volume101 ret = volume_start(self.mnode, self.volname, force=True)102 self.assertTrue(ret, "Error in force start the volume")103 g.log.info('Volume restart success')104 sleep(10)105 # Delete parent_dir106 ret = rmdir(self.clients[0], parent_dir, force=True)107 self.assertTrue(ret, ('rmdir failed for %s ' % parent_dir))108 g.log.info("rmdir of directory %s successful", parent_dir)109 def test_rmdir_dir_when_hash_nonhash_vol_down(self):110 """111 case -2:112 - create dir1 and dir2113 - bring down hashed subvol for dir1114 - bring down a non-hashed subvol for dir2115 - rmdir dir1 should fail with ENOTCONN116 - rmdir dir2 should fail with ENOTCONN117 """118 # pylint: disable=protected-access119 # pylint: disable=too-many-statements120 # pylint: disable=unsubscriptable-object121 # Create dir1 and dir2122 directory_list = []123 for number in range(1, 3):124 directory_list.append('{}/dir{}'.format(self.mountpoint, number))125 ret = mkdir(self.clients[0], directory_list[-1])126 self.assertTrue(ret, ('mkdir failed for %s '127 % directory_list[-1]))128 g.log.info("mkdir of directory %s successful",129 directory_list[-1])130 # Find a non hashed subvolume(or brick)131 nonhashed_subvol, count = find_nonhashed_subvol(self.subvols, "/",132 "dir1")133 self.assertIsNotNone(nonhashed_subvol,134 "Error in finding nonhashed value")135 g.log.info("nonhashed_subvol %s", nonhashed_subvol._host)136 # Bring nonhashed_subbvol offline137 ret = bring_bricks_offline(self.volname, self.subvols[count])138 self.assertTrue(ret, ('Error in bringing down subvolume %s'139 % self.subvols[count]))140 g.log.info('target subvol %s is offline', self.subvols[count])141 # 'rmdir' on dir1 should fail with ENOTCONN142 ret = rmdir(self.clients[0], directory_list[0])143 self.assertFalse(ret, ('Expected rmdir to fail for %s'144 % directory_list[0]))145 g.log.info("rmdir of directory %s failed as expected",146 directory_list[0])147 # Bring up the subvol - restart volume148 ret = volume_start(self.mnode, self.volname, force=True)149 self.assertTrue(ret, "Error in force start the volume")150 g.log.info('Volume restart success')151 sleep(10)152 # Unmounting and Mounting the volume back to Heal153 ret, _, err = umount_volume(self.clients[1], self.mountpoint)154 self.assertFalse(ret, "Error in creating temp mount %s" % err)155 ret, _, err = mount_volume(self.volname,156 mtype='glusterfs',157 mpoint=self.mountpoint,158 mserver=self.servers[0],159 mclient=self.clients[1])160 self.assertFalse(ret, "Error in creating temp mount")161 ret, _, _ = g.run(self.clients[1], ("ls %s/dir1" % self.mountpoint))162 self.assertEqual(ret, 0, "Error in lookup for dir1")163 g.log.info("lookup successful for dir1")164 # This confirms that healing is done on dir1165 ret = validate_files_in_dir(self.clients[0],166 directory_list[0],167 test_type=LAYOUT_IS_COMPLETE,168 file_type=FILETYPE_DIRS)169 self.assertTrue(ret, "validate_files_in_dir for dir1 failed")170 g.log.info("healing successful for dir1")171 # Bring down the hashed subvol172 # Find a hashed subvolume(or brick)173 hashed_subvol, count = find_hashed_subvol(self.subvols, "/", "dir2")174 self.assertIsNotNone(hashed_subvol,175 "Error in finding nonhashed value")176 g.log.info("hashed_subvol %s", hashed_subvol._host)177 # Bring hashed_subbvol offline178 ret = bring_bricks_offline(self.volname, self.subvols[count])179 self.assertTrue(ret, ('Error in bringing down subvolume %s',180 self.subvols[count]))181 g.log.info('target subvol %s is offline', self.subvols[count])182 # 'rmdir' on dir2 should fail with ENOTCONN183 ret = rmdir(self.clients[0], directory_list[1])184 self.assertFalse(ret, ('Expected rmdir to fail for %s'185 % directory_list[1]))186 g.log.info("rmdir of dir2 directory %s failed as expected",187 directory_list[1])188 # Cleanup189 # Bring up the subvol - restart the volume190 ret = volume_start(self.mnode, self.volname, force=True)191 self.assertTrue(ret, "Error in force start the volume")192 g.log.info('Volume restart success')193 sleep(10)194 # Delete dirs195 for directory in directory_list:196 ret = rmdir(self.clients[0], directory)197 self.assertTrue(ret, ('rmdir failed for %s ' % directory))198 g.log.info("rmdir of directory %s successful", directory)199 def test_rm_file_when_nonhash_vol_down(self):200 """201 case -3:202 - create parent203 - mkdir parent/child204 - touch parent/child/file205 - bringdown a subvol where file is not present206 - rm -rf parent207 - Only file should be deleted208 - rm -rf of parent should fail with ENOTCONN209 """210 # pylint: disable=protected-access211 # pylint: disable=too-many-statements212 # pylint: disable=unsubscriptable-object213 # Find a non hashed subvolume(or brick)214 # Create parent dir215 parent_dir = self.mountpoint + '/parent'216 child_dir = parent_dir + '/child'217 ret = mkdir(self.clients[0], parent_dir)218 self.assertTrue(ret, ('mkdir failed for %s ' % parent_dir))219 g.log.info("mkdir of parent directory %s successful", parent_dir)220 # Create child dir221 ret = mkdir(self.clients[0], child_dir)222 self.assertTrue(ret, ('mkdir failed for %s ' % child_dir))223 g.log.info("mkdir of child directory %s successful", child_dir)224 # Create a file under child_dir225 file_one = child_dir + '/file_one'226 ret, _, err = g.run(self.clients[0], ("touch %s" % file_one))227 self.assertFalse(ret, ('touch failed for %s err: %s' %228 (file_one, err)))229 # Find a non hashed subvolume(or brick)230 nonhashed_subvol, count = find_nonhashed_subvol(self.subvols,231 "parent/child",232 "file_one")233 self.assertIsNotNone(nonhashed_subvol,234 "Error in finding nonhashed value")235 g.log.info("nonhashed_subvol %s", nonhashed_subvol._host)236 # Bring nonhashed_subbvol offline237 ret = bring_bricks_offline(self.volname, self.subvols[count])238 self.assertTrue(ret, ('Error in bringing down subvolume %s'239 % self.subvols[count]))240 g.log.info('target subvol %s is offline', self.subvols[count])241 # 'rm -rf' on parent should fail with ENOTCONN242 ret = rmdir(self.clients[0], parent_dir)243 self.assertFalse(ret, ('Expected rmdir to fail for %s' % parent_dir))244 g.log.info("rmdir of parent directory %s failed as expected"245 " with err %s", parent_dir, err)246 brickobject = create_brickobjectlist(self.subvols, "parent/child")247 self.assertIsNotNone(brickobject,248 "could not create brickobject list")249 # Make sure file_one is deleted250 for brickdir in brickobject:251 dir_path = "%s/parent/child/file_one" % brickdir.path252 brick_path = dir_path.split(":")253 self.assertTrue((file_exists(brickdir._host, brick_path[1])) == 0,254 ('Expected file %s not to exist on servers'255 % parent_dir))256 g.log.info("file is deleted as expected")257 # Cleanup258 # Bring up the subvol - restart volume259 ret = volume_start(self.mnode, self.volname, force=True)260 self.assertTrue(ret, "Error in force start the volume")261 g.log.info('Volume restart success.')262 sleep(10)263 # Delete parent_dir264 ret = rmdir(self.clients[0], parent_dir, force=True)265 self.assertTrue(ret, ('rmdir failed for %s ' % parent_dir))266 g.log.info("rmdir of directory %s successful", parent_dir)267 def test_rmdir_parent_pre_nonhash_vol_down(self):268 """269 case -4:270 - Bring down a non-hashed subvol for parent_dir271 - mkdir parent272 - rmdir parent should fails with ENOTCONN273 """274 # pylint: disable=protected-access275 # pylint: disable=too-many-statements276 # pylint: disable=unsubscriptable-object277 nonhashed_subvol, count = find_nonhashed_subvol(self.subvols,278 "/", "parent")279 self.assertIsNotNone(nonhashed_subvol,280 'Error in finding nonhashed subvol')281 g.log.info("nonhashed subvol %s", nonhashed_subvol._host)282 # Bring nonhashed_subbvol offline283 ret = bring_bricks_offline(self.volname, self.subvols[count])284 self.assertTrue(ret, ('Error in bringing down subvolume %s'285 % self.subvols[count]))286 g.log.info('target subvol %s is offline', self.subvols[count])287 parent_dir = self.mountpoint + '/parent'288 ret = mkdir(self.clients[0], parent_dir)289 self.assertTrue(ret, ('mkdir failed for %s ' % parent_dir))290 g.log.info("mkdir of parent directory %s successful", parent_dir)291 # 'rmdir' on parent should fail with ENOTCONN292 ret = rmdir(self.clients[0], parent_dir)293 self.assertFalse(ret, ('Expected rmdir to fail for %s' % parent_dir))294 g.log.info("rmdir of parent directory %s failed as expected",295 parent_dir)296 # Cleanup297 # Bring up the subvol - restart volume298 ret = volume_start(self.mnode, self.volname, force=True)299 self.assertTrue(ret, "Error in force start the volume")300 g.log.info('Volume restart success.')301 sleep(10)302 # Delete parent_dir303 ret = rmdir(self.clients[0], parent_dir, force=True)304 self.assertTrue(ret, ('rmdir failed for %s ' % parent_dir))305 g.log.info("rmdir of directory %s successful", parent_dir)306 def tearDown(self):307 """308 Unmount Volume and Cleanup Volume309 """310 ret = self.unmount_volume_and_cleanup_volume(mounts=self.mounts)311 if not ret:312 raise ExecutionError("Failed to Unmount Volume and Cleanup Volume")313 g.log.info("Successful in Unmount Volume and Cleanup Volume")314 # Calling GlusterBaseClass tearDown...
rmdir.py
Source:rmdir.py
...7# Remove a directory that does not exist in the lower layer8def subtest_1(ctx):9 """Remove nonexistent directory"""10 d = ctx.no_dir() + ctx.termslash()11 ctx.rmdir(d, err=ENOENT)12 ctx.rmdir(d, err=ENOENT)13# Remove a subdirectory from a dir that does not exist14def subtest_2(ctx):15 """Remove subdir from nonexistent directory"""16 d = ctx.no_dir() + "/sub" + ctx.termslash()17 ctx.rmdir(d, err=ENOENT)18 ctx.rmdir(d, err=ENOENT)19# Rmdir a file20def subtest_3(ctx):21 """Remove-dir a file"""22 f = ctx.reg_file()23 d = ctx.reg_file() + ctx.termslash()24 ctx.rmdir(d, err=ENOTDIR)25 ctx.rmdir(d, err=ENOTDIR)26 ctx.open_file(f, ro=1, read=":xxx:yyy:zzz")27# Remove a subdir from a file28def subtest_4(ctx):29 """Remove subdir from file"""30 f = ctx.reg_file()31 d = ctx.reg_file() + "/sub" + ctx.termslash()32 ctx.rmdir(d, err=ENOTDIR)33 ctx.rmdir(d, err=ENOTDIR)34 ctx.open_file(f, ro=1, read=":xxx:yyy:zzz")35# Remove an empty lower directory36def subtest_5(ctx):37 """Remove empty dir"""38 d = ctx.empty_dir() + ctx.termslash()39 subdir = d + "/sub" + ctx.termslash()40 ctx.rmdir(d)41 ctx.rmdir(d, err=ENOENT)42 ctx.rmdir(subdir, err=ENOENT)43# Remove a non-existent directory from an empty lower directory44def subtest_6(ctx):45 """Remove directory from empty dir"""46 d = ctx.empty_dir() + "/sub" + ctx.termslash()47 ctx.rmdir(d, err=ENOENT)48 ctx.rmdir(d, err=ENOENT)49# Remove a populated lower directory50def subtest_7(ctx):51 """Remove populated directory"""52 d = ctx.non_empty_dir() + ctx.termslash()53 f = d + "/a"54 ctx.rmdir(d, err=ENOTEMPTY)55 ctx.open_file(f, ro=1, read="")56 ctx.unlink(f)57 ctx.open_file(f, ro=1, err=ENOENT)58 ctx.unlink(f, err=ENOENT)59 ctx.rmtree(d)60 ctx.open_file(f, ro=1, read="", err=ENOENT)61# Remove a populated lower directory after creating a file in it62def subtest_8(ctx):63 """Remove populated directory with created file"""64 d = ctx.empty_dir() + ctx.termslash()65 f = d + "/b"66 ctx.open_file(f, wo=1, crt=1, ex=1, write="abcq")67 ctx.rmdir(d, err=ENOTEMPTY)68 ctx.unlink(f)69 ctx.open_file(f, ro=1, err=ENOENT)70 ctx.unlink(f, err=ENOENT)71 ctx.rmtree(d)72 ctx.open_file(f, ro=1, read="", err=ENOENT)73# Remove a populated lower directory with copied-up file74def subtest_9(ctx):75 """Remove populated directory with copied up file"""76 d = ctx.non_empty_dir() + ctx.termslash()77 f = d + "/a"78 ctx.rmdir(d, err=ENOTEMPTY)79 ctx.open_file(f, ro=1, read="")80 ctx.open_file(f, wo=1, write="abcd")81 ctx.open_file(f, ro=1, read="abcd")82 ctx.unlink(f)83 ctx.open_file(f, ro=1, err=ENOENT)84 ctx.unlink(f, err=ENOENT)85 ctx.rmtree(d)86 ctx.open_file(f, ro=1, read="", err=ENOENT)87# Remove a populated lower directory after unlinking a file and creating a dir over it88def subtest_10(ctx):89 """Remove populated directory with mkdir after unlink"""90 d = ctx.non_empty_dir() + ctx.termslash()91 f = d + "/a"92 ctx.rmdir(d, err=ENOTEMPTY)93 ctx.open_file(f, ro=1, read="")94 ctx.rmdir(d, err=ENOTEMPTY)95 ctx.unlink(f)96 ctx.open_file(f, ro=1, err=ENOENT)97 ctx.unlink(f, err=ENOENT)98 ctx.mkdir(f, 0o755)99 ctx.mkdir(f, 0o755, err=EEXIST)100 ctx.rmdir(d, err=ENOTEMPTY)101 ctx.rmdir(f)102 ctx.rmdir(f, err=ENOENT)103 ctx.rmtree(d)104 ctx.open_file(f, ro=1, read="", err=ENOENT)105# Remove a directory from a populated lower directory and recreate it106def subtest_11(ctx):107 """Remove directory from dir"""108 d = ctx.non_empty_dir() + ctx.termslash()109 pop = d + "/pop"110 subdir = pop + "/c" + ctx.termslash()111 f = pop + "/b"112 ctx.rmdir(subdir)113 ctx.rmdir(subdir, err=ENOENT)114 ctx.mkdir(subdir, 0o755)115 ctx.mkdir(subdir, 0o755, err=EEXIST)116 ctx.open_file(f, ro=1, read=":aaa:bbb:ccc")117 ctx.rmtree(d)118 ctx.open_file(f, ro=1, err=ENOENT)119# Remove directory symlinks pointing to a file120def subtest_12(ctx):121 """Remove-dir symlinks to file"""122 f = ctx.reg_file()123 d = ctx.reg_file() + ctx.termslash()124 sym = ctx.direct_sym() + ctx.termslash()125 isym = ctx.indirect_sym() + ctx.termslash()126 ctx.rmdir(isym, err=ENOTDIR)127 ctx.rmdir(isym, err=ENOTDIR)128 ctx.rmdir(sym, err=ENOTDIR)129 ctx.rmdir(sym, err=ENOTDIR)130 ctx.rmdir(d, err=ENOTDIR)131 ctx.rmdir(d, err=ENOTDIR)132 ctx.open_file(f, ro=1, read=":xxx:yyy:zzz")133# Remove a directory over a symlink to a dir134def subtest_13(ctx):135 """Remove directory over sym to dir"""136 d = ctx.non_empty_dir() + ctx.termslash()137 sym = ctx.direct_dir_sym() + ctx.termslash()138 ctx.rmdir(sym, err=ENOTDIR)139 ctx.rmdir(sym, err=ENOTDIR)140 ctx.rmdir(d, err=ENOTEMPTY)141 ctx.open_file(sym + "/a", ro=1, read="")142 ctx.open_file(d + "/a", ro=1, read="")143 ctx.rmtree(d)144 ctx.open_file(sym + "/a", ro=1, err=ENOENT)145 ctx.open_file(d + "/a", ro=1, err=ENOENT)146# Remove a directory over a symlink to a symlink to a dir147def subtest_14(ctx):148 """Remove directory over sym to sym to dir"""149 d = ctx.non_empty_dir() + ctx.termslash()150 sym = ctx.direct_dir_sym() + ctx.termslash()151 isym = ctx.indirect_dir_sym() + ctx.termslash()152 ctx.rmdir(isym, err=ENOTDIR)153 ctx.rmdir(isym, err=ENOTDIR)154 ctx.rmdir(sym, err=ENOTDIR)155 ctx.rmdir(d, err=ENOTEMPTY)156 ctx.open_file(isym + "/a", ro=1, read="")157 ctx.open_file(sym + "/a", ro=1, read="")158 ctx.open_file(d + "/a", ro=1, read="")159 ctx.rmtree(d)160 ctx.open_file(isym + "/a", ro=1, err=ENOENT)161 ctx.open_file(sym + "/a", ro=1, err=ENOENT)162 ctx.open_file(d + "/a", ro=1, err=ENOENT)163# Remove a directory over a dangling symlink164def subtest_15(ctx):165 """Remove directory over dangling sym"""166 d = ctx.no_file() + ctx.termslash()167 sym = ctx.pointless() + ctx.termslash()168 ctx.rmdir(sym, err=ENOTDIR)169 ctx.rmdir(sym, err=ENOTDIR)170 ctx.rmdir(d, err=ENOENT)171# Remove an empty lower directory, recreate, populate and try to remove172def subtest_16(ctx):173 """Remove non-empty opaque directory"""174 d = ctx.empty_dir() + ctx.termslash()175 f = d + "/b"176 ctx.rmdir(d)177 ctx.rmdir(d, err=ENOENT)178 ctx.mkdir(d, 0o755)179 ctx.open_file(f, wo=1, crt=1, ex=1, write="abcq")180 ctx.rmdir(d, err=ENOTEMPTY)181 ctx.unlink(f)182 ctx.open_file(f, ro=1, err=ENOENT)183 ctx.unlink(f, err=ENOENT)...
__init__.py
Source:__init__.py
1# uncompyle6 version 2.9.102# Python bytecode 2.7 (62211)3# Decompiled from: Python 3.6.0b2 (default, Oct 11 2016, 05:27:10) 4# [GCC 6.2.0 20161005]5# Embedded file name: __init__.py6import dsz7import dsz.cmd8import dsz.data9import dsz.lp10class RmDir(dsz.data.Task):11 def __init__(self, cmd=None):12 dsz.data.Task.__init__(self, cmd)13 def _LoadData(self):14 try:15 self.TargetDirectory = RmDir.TargetDirectory(dsz.cmd.data.Get('TargetDirectory', dsz.TYPE_OBJECT)[0])16 except:17 self.TargetDirectory = None18 return19 class TargetDirectory(dsz.data.DataBean):20 def __init__(self, obj):21 try:22 self.path = dsz.cmd.data.ObjectGet(obj, 'path', dsz.TYPE_STRING)[0]23 except:24 self.path = None25 return26dsz.data.RegisterCommand('RmDir', RmDir)27RMDIR = RmDir...
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?
