KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use mkfs_exec method in autotest

Best Python code snippet using autotest_python

mount_test.py

Source:mount_test.py

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 vdsm.storage 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.seek(size)29        f.write('\0')30    try:31        commands.run([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            commands.run([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 = "foobar.baz.qux.com:/var/lib/exports/%04d"218        mountpoint = ("/rhev/data-center/mnt/foobar.baz.qux.com:_var_lib"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=10.35.0.102,mountvers=3,mountport=892,"223                   "mountproto=udp,local_lock=none,addr=10.35.0.102")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...

__init__.py

Source:__init__.py

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' % self.name):43            try:44                coreutils.modprobe(self.name)45            except:46                e = sys.exc_info()[1]47                error_text="Cannot load '%s' kernel module: %s" % (self.name, 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', self.name]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 = re.search('%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)...

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.