How to use mkfs_exec method in autotest

Best Python code snippet using autotest_python Github


Full Screen

1# SPDX-FileCopyrightText: Red Hat, Inc.2# SPDX-License-Identifier: GPL-2.0-or-later3from __future__ import absolute_import4from __future__ import division5from __future__ import print_function6from contextlib import contextmanager7import errno8from tempfile import mkstemp9import os10import time11import pytest12from vdsm.common.units import MiB, GiB13from vdsm.common import commands14from import mount15from nose.plugins.skip import SkipTest16from testlib import VdsmTestCase17from testlib import namedTemporaryDir, temporaryPath18from testlib import expandPermutations, permutations19from testValidation import broken_on_ci20import monkeypatch21from . marks import requires_root22FLOPPY_SIZE = 4 * MiB23MKFS_EXEC = '/usr/sbin/mkfs.ext2'24@contextmanager25def createFloppyImage(size):26 fd, path = mkstemp()27 with os.fdopen(fd, "w") as f:28 f.write('\0')30 try:31[MKFS_EXEC, "-F", path])32 except OSError as e:33 if e.errno == errno.ENOENT:34 raise SkipTest("cannot execute " + MKFS_EXEC)35 raise36 try:37 yield path38 finally:39 os.unlink(path)40@expandPermutations41class TestMountEquality(VdsmTestCase):42 def test_eq_equal(self):43 m1 = mount.Mount("spec", "file")44 m2 = mount.Mount("spec", "file")45 self.assertTrue(m1 == m2, "%s should equal %s" % (m1, m2))46 def test_eq_subclass(self):47 class Subclass(mount.Mount):48 pass49 m1 = mount.Mount("spec", "file")50 m2 = Subclass("spec", "file")51 self.assertFalse(m1 == m2, "%s should not equal %s" % (m1, m2))52 @permutations([53 ("spec", "spec", "file1", "file2"),54 ("spec1", "spec2", "file", "file"),55 ])56 def test_eq_different(self, spec1, spec2, file1, file2):57 m1 = mount.Mount(spec1, file1)58 m2 = mount.Mount(spec2, file2)59 self.assertFalse(m1 == m2, "%s should not equal %s" % (m1, m2))60 def test_ne_equal(self):61 m1 = mount.Mount("spec", "file")62 m2 = mount.Mount("spec", "file")63 self.assertFalse(m1 != m2, "%s should equal %s" % (m1, m2))64@expandPermutations65class TestMountHash(VdsmTestCase):66 def test_equal_same_hash(self):67 m1 = mount.Mount("spec", "file")68 m2 = mount.Mount("spec", "file")69 self.assertEqual(hash(m1), hash(m2))70 def test_subclass_different_hash(self):71 class Subclass(mount.Mount):72 pass73 m1 = mount.Mount("spec", "file")74 m2 = Subclass("spec", "file")75 self.assertNotEqual(hash(m1), hash(m2))76 @permutations([77 ("spec", "spec", "file1", "file2"),78 ("spec1", "spec2", "file", "file"),79 ])80 def test_not_equal_different_hash(self, spec1, spec2, file1, file2):81 m1 = mount.Mount(spec1, file1)82 m2 = mount.Mount(spec2, file2)83 self.assertNotEqual(hash(m1), hash(m2))84@contextmanager85def loop_mount(m):86 m.mount(mntOpts="loop")87 try:88 yield89 finally:90 time.sleep(0.5)91 m.umount()92@expandPermutations93class TestMount(VdsmTestCase):94 @requires_root95 @pytest.mark.root96 @broken_on_ci("mount check fails after successful mount", name="TRAVIS_CI")97 def testLoopMount(self):98 with namedTemporaryDir() as mpath:99 # two nested with blocks to be python 2.6 friendly100 with createFloppyImage(FLOPPY_SIZE) as path:101 m = mount.Mount(path, mpath)102 with loop_mount(m):103 self.assertTrue(m.isMounted())104 @requires_root105 @pytest.mark.root106 @broken_on_ci("mount check fails after successful mount", name="TRAVIS_CI")107 def testSymlinkMount(self):108 with namedTemporaryDir() as root_dir:109 backing_image = os.path.join(root_dir, 'backing.img')110 link_to_image = os.path.join(root_dir, 'link_to_image')111 mountpoint = os.path.join(root_dir, 'mountpoint')112 with open(backing_image, 'w') as f:113 os.ftruncate(f.fileno(), GiB)114[MKFS_EXEC, "-F", backing_image])115 os.symlink(backing_image, link_to_image)116 os.mkdir(mountpoint)117 m = mount.Mount(link_to_image, mountpoint)118 with loop_mount(m):119 self.assertTrue(m.isMounted())120 @permutations([121 # Only fs_spec matches122 ("server:/path", "/mnt/server:_other__path", False),123 # Only fs_file matches124 ("server:/other_path", "/mnt/server:_path", False),125 # Both fs_spec and fs_file don't match126 ("server:/other_path", "/mnt/server:_other__path", False),127 # Both match128 ("server:/path", "/mnt/server:_path", True),129 ])130 def test_is_mounted(self, fs_spec, fs_file, equality):131 """132 Verifies that both fs_spec and fs_file match the mounted target.133 """134 with fake_mounts(["server:/path /mnt/server:_path nfs defaults 0 0"]):135 mnt = mount.Mount(fs_spec, fs_file)136 self.assertEqual(mnt.isMounted(), equality)137 @permutations([138 # NFS4 using fsid=0 - kernel display mount as server://path instead of139 # normalized server:/path140 ("server://a/b /mnt/server:_a_b nfs defaults 0 0",),141 # Not seen yet, but it should work now142 ("server:/a//b /mnt/server:_a_b nfs defaults 0 0",),143 ("server:/a/b// /mnt/server:_a_b nfs defaults 0 0",),144 ])145 def test_is_mounted_normalize_kernel_mounts(self, mount_line):146 with fake_mounts([mount_line]):147 mnt = mount.Mount("server:/a/b", "/mnt/server:_a_b")148 self.assertTrue(mnt.isMounted())149 def test_is_mounted_with_symlink(self):150 with namedTemporaryDir() as dir:151 file = os.path.join(dir, "file")152 open(file, "w").close()153 link_to_file = os.path.join(dir, "link_to_file")154 os.symlink(file, link_to_file)155 mountpoint = "/mnt/mountpoint"156 with fake_mounts(["%s %s nfs defaults 0 0" %157 (link_to_file, mountpoint)]):158 mnt = mount.Mount(link_to_file, mountpoint)159 self.assertTrue(mnt.isMounted())160 def test_is_mounted_gluster_with_rdma(self):161 with fake_mounts(162 ["server:/volume.rdma /mnt/server:volume fuse.glusterfs "163 "defaults 0 0"]):164 mnt = mount.Mount("server:/volume", "/mnt/server:volume")165 self.assertTrue(mnt.isMounted())166@contextmanager167def fake_mounts(mount_lines):168 """169 This method gets a list of mount lines,170 fakes the /proc/mounts and /etc/mtab files171 using monkey patch with a temporary file,172 and cleans everything on the end of use.173 Usage example:174 with fake_mounts([mount_line_1, mount_line_2]):175 <do something with /proc/mounts or /etc/mtab>176 """177 data = "".join(line + "\n" for line in mount_lines)178 with temporaryPath(data=data.encode("utf-8")) as fake_mounts:179 with monkeypatch.MonkeyPatchScope([180 (mount, '_PROC_MOUNTS_PATH', fake_mounts),181 ]):182 yield183class TestRemoteSdIsMounted(VdsmTestCase):184 def test_is_mounted(self):185 with fake_mounts(["server:/path "186 "/rhev/data-center/mnt/server:_path "187 "nfs4 defaults 0 0"]):188 self.assertTrue(mount.isMounted(189 "/rhev/data-center/mnt/server:_path"))190 def test_is_mounted_deleted(self):191 with fake_mounts([u"server:/path "192 u"/rhev/data-center/mnt/server:_path\\040(deleted) "193 u"nfs4 defaults 0 0"]):194 self.assertTrue(mount.isMounted(195 "/rhev/data-center/mnt/server:_path"))196 def test_path_with_spaces(self):197 with fake_mounts(198 [u"server:/a\\040b /mnt/server:_a\\040b nfs4 opts 0 0"]):199 self.assertTrue(mount.isMounted("/mnt/server:_a b"))200 self.assertFalse(mount.isMounted(u"/mnt/server:_a\\040b"))201 def test_path_with_backslash(self):202 with fake_mounts(203 [u"server:/a\\134040b /mnt/server:_a\\134040b nfs4 opts 0 0"]):204 self.assertTrue(mount.isMounted(u"/mnt/server:_a\\040b"))205 self.assertFalse(mount.isMounted(u"/mnt/server:_a\\134040b"))206 def test_is_not_mounted(self):207 with fake_mounts(["server:/path "208 "/rhev/data-center/mnt/server:_path "209 "nfs4 defaults 0 0"]):210 self.assertFalse(mount.isMounted(211 "/rhev/data-center/mnt/server:_other_path"))212@expandPermutations213class TestIsMountedTiming(VdsmTestCase):214 @pytest.mark.stress215 @permutations([[1], [50], [100], [1000]])216 def test_is_mounted(self, count):217 server = ""218 mountpoint = ("/rhev/data-center/mnt/"219 "_exports_%04d")220 options = ("rw,relatime,vers=3,rsize=524288,wsize=524288,namlen=255,"221 "soft,nosharecache,proto=tcp,timeo=600,retrans=6,sec=sys,"222 "mountaddr=,mountvers=3,mountport=892,"223 "mountproto=udp,local_lock=none,addr=")224 version = "nfs"225 freq = "0"226 passno = "0"227 lines = []228 for i in range(count):229 line = " ".join((server % i, mountpoint % i, options, version,230 freq, passno))231 lines.append(line)232 with fake_mounts(lines):233 start = time.time()234 self.assertTrue(mount.isMounted(mountpoint % i))235 elapsed = time.time() - start...

Full Screen

Full Screen Github


Full Screen

1from __future__ import with_statement2'''3Created on Nov 11, 20104@author: spike5@author: marat6'''7from .. import MKFS_EXEC, MOUNT_EXEC8from scalarizr.linux import coreutils, pkgmgr9from scalarizr.util import system2, PopenError10import os11import re12import sys13import logging14logger = logging.getLogger(__name__)15class FileSystemError(PopenError):16 pass17def system(*args, **kwargs):18 kwargs['logger'] = logger19 kwargs['exc_class'] = FileSystemError20 kwargs['warn_stderr'] = False21 return system2(*args, **kwargs)22def device_should_exists(f):23 def d(*args):24 if not os.path.exists(args[1]):25 raise FileSystemError("Device %s doesn't exist" % args[1])26 return f(*args)27 return d28class FileSystem:29 name = None30 freezable = False31 resizable = True32 umount_on_resize = False33 os_packages = None34 E_MKFS = 'Error during filesystem creation on device %s'35 E_RESIZE = 'Error during filesystem resize on device %s'36 E_SET_LABEL = 'Error while setting label for device %s'37 E_GET_LABEL = 'Error while getting label for device %s'38 E_FREEZE = 'Error during filesystem freeze on device %s'39 E_UNFREEZE = 'Error during filesystem un-freeze on device %s'40 E_NOT_MOUNTED = 'Device %s should be mounted'41 def __init__(self):42 if not os.path.exists('/sbin/mkfs.%s' % try:44 coreutils.modprobe( except:46 e = sys.exc_info()[1]47 error_text="Cannot load '%s' kernel module: %s" % (, e)48 raise Exception(error_text)49 if self.os_packages:50 for package in self.os_packages:51 pkgmgr.installed(package)52 @device_should_exists53 def mkfs(self, device, options=None):54 cmd = [MKFS_EXEC, '-t',]55 if options:56 cmd.extend(options)57 cmd.append(device)58 system(cmd, error_text=self.E_MKFS % device)59 def resize(self, device, size=None, **options):60 '''61 Resize filesystem on given device to given size (default: to the size of partition)62 '''63 pass64 def set_label(self, device, label):65 pass66 def get_label(self, device):67 pass68 def freeze(self, device):69 pass70 def unfreeze(self, device):71 pass72 def _check_mounted(self, device):73 res ='%s\s+on\s+(?P<mpoint>.+)\s+type' % device, system(MOUNT_EXEC)[0])74 if not res:75 raise FileSystemError(self.E_NOT_MOUNTED % device)...

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?