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

How to use cleanup_env method in lisa

Best Python code snippet using lisa_python

virsh_pool_acl.py

Source:virsh_pool_acl.py

1import re2import os3import logging4from autotest.client import utils5from autotest.client import lv_utils6from autotest.client.shared import error7from virttest import libvirt_storage8from virttest import utils_test9from virttest import virsh10from virttest.libvirt_xml.pool_xml import PoolXML11from provider import libvirt_version12def run(test, params, env):13    """14    Test the virsh pool commands with acl, initiate a pool then do15    following operations.16    (1) Undefine a given type pool17    (2) Define the pool from xml18    (3) Build given type pool19    (4) Start pool20    (5) Destroy pool21    (6) Refresh pool after start it22    (7) Run vol-list with the pool23    (9) Delete pool24    For negative cases, redo failed step to make the case run continue.25    Run cleanup at last restore env.26    """27    # Initialize the variables28    pool_name = params.get("pool_name", "temp_pool_1")29    pool_type = params.get("pool_type", "dir")30    pool_target = params.get("pool_target", "")31    # The file for dumped pool xml32    pool_xml = os.path.join(test.tmpdir, "pool.xml.tmp")33    if os.path.dirname(pool_target) is "":34        pool_target = os.path.join(test.tmpdir, pool_target)35    vol_name = params.get("vol_name", "temp_vol_1")36    # Use pool name as VG name37    vg_name = pool_name38    vol_path = os.path.join(pool_target, vol_name)39    define_acl = "yes" == params.get("define_acl", "no")40    undefine_acl = "yes" == params.get("undefine_acl", "no")41    start_acl = "yes" == params.get("start_acl", "no")42    destroy_acl = "yes" == params.get("destroy_acl", "no")43    build_acl = "yes" == params.get("build_acl", "no")44    delete_acl = "yes" == params.get("delete_acl", "no")45    refresh_acl = "yes" == params.get("refresh_acl", "no")46    vol_list_acl = "yes" == params.get("vol_list_acl", "no")47    list_dumpxml_acl = "yes" == params.get("list_dumpxml_acl", "no")48    src_pool_error = "yes" == params.get("src_pool_error", "no")49    define_error = "yes" == params.get("define_error", "no")50    undefine_error = "yes" == params.get("undefine_error", "no")51    start_error = "yes" == params.get("start_error", "no")52    destroy_error = "yes" == params.get("destroy_error", "no")53    build_error = "yes" == params.get("build_error", "no")54    delete_error = "yes" == params.get("delete_error", "no")55    refresh_error = "yes" == params.get("refresh_error", "no")56    vol_list_error = "yes" == params.get("vol_list_error", "no")57    dir_mode_check = "yes" == params.get("dir_mode_check", "no")58    dir_original_mode = params.get("dir_original_mode")59    if dir_mode_check:60        if not os.path.exists(pool_target):61            os.mkdir(pool_target)62        utils.run("chmod %s %s" % (dir_original_mode, pool_target))63    # Clean up flags:64    # cleanup_env[0] for nfs, cleanup_env[1] for iscsi, cleanup_env[2] for lvm65    # cleanup_env[3] for selinux backup status, cleanup_env[4] for gluster66    cleanup_env = [False, False, False, "", False]67    # libvirt acl related params68    uri = params.get("virsh_uri")69    unprivileged_user = params.get('unprivileged_user')70    if unprivileged_user:71        if unprivileged_user.count('EXAMPLE'):72            unprivileged_user = 'testacl'73    if not libvirt_version.version_compare(1, 1, 1):74        if params.get('setup_libvirt_polkit') == 'yes':75            raise error.TestNAError("API acl test not supported in current"76                                    " libvirt version.")77    acl_dargs = {'uri': uri, 'unprivileged_user': unprivileged_user,78                 'debug': True}79    def check_exit_status(result, expect_error=False):80        """81        Check the exit status of virsh commands.82        :param result: Virsh command result object83        :param expect_error: Boolean value, expect command success or fail84        """85        if not expect_error:86            if result.exit_status != 0:87                raise error.TestFail(result.stderr)88            else:89                logging.debug("Command output:\n%s", result.stdout.strip())90        elif expect_error and result.exit_status == 0:91            raise error.TestFail("Expect fail, but run successfully.")92    def check_pool_list(pool_name, option="--all", expect_error=False):93        """94        Check pool by running pool-list command with given option.95        :param pool_name: Name of the pool96        :param option: option for pool-list command97        :param expect_error: Boolean value, expect command success or fail98        """99        found = False100        # Get the list stored in a variable101        if list_dumpxml_acl:102            result = virsh.pool_list(option, **acl_dargs)103        else:104            result = virsh.pool_list(option, ignore_status=True)105        check_exit_status(result, False)106        output = re.findall(r"(\S+)\ +(\S+)\ +(\S+)[\ +\n]",107                            str(result.stdout))108        for item in output:109            if pool_name in item[0]:110                found = True111                break112        if found:113            logging.debug("Find pool '%s' in pool list.", pool_name)114        else:115            logging.debug("Not find pool %s in pool list.", pool_name)116        if expect_error and found:117            raise error.TestFail("Unexpect pool '%s' exist." % pool_name)118        if not expect_error and not found:119            raise error.TestFail("Expect pool '%s' doesn't exist." % pool_name)120    # Run Testcase121    try:122        _pool = libvirt_storage.StoragePool()123        # Init a pool for test124        result = utils_test.libvirt.define_pool(pool_name, pool_type,125                                                pool_target, cleanup_env)126        check_exit_status(result, src_pool_error)127        option = "--inactive --type %s" % pool_type128        check_pool_list(pool_name, option)129        if list_dumpxml_acl:130            xml = virsh.pool_dumpxml(pool_name, to_file=pool_xml, **acl_dargs)131        else:132            xml = virsh.pool_dumpxml(pool_name, to_file=pool_xml)133        logging.debug("Pool '%s' XML:\n%s", pool_name, xml)134        # Step (1)135        # Undefine pool136        if undefine_acl:137            result = virsh.pool_undefine(pool_name, **acl_dargs)138        else:139            result = virsh.pool_undefine(pool_name, ignore_status=True)140        check_exit_status(result, undefine_error)141        if undefine_error:142            check_pool_list(pool_name, "--all", False)143            # Redo under negative case to keep case continue144            result = virsh.pool_undefine(pool_name, ignore_status=True)145            check_exit_status(result)146            check_pool_list(pool_name, "--all", True)147        else:148            check_pool_list(pool_name, "--all", True)149        # Step (2)150        # Define pool from XML file151        if define_acl:152            result = virsh.pool_define(pool_xml, **acl_dargs)153        else:154            result = virsh.pool_define(pool_xml)155        check_exit_status(result, define_error)156        if define_error:157            # Redo under negative case to keep case continue158            result = virsh.pool_define(pool_xml)159            check_exit_status(result)160        # Step (3)161        # Buid pool, this step may fail for 'disk' and 'logical' types pool162        if pool_type not in ["disk", "logical"]:163            option = ""164        # Options --overwrite and --no-overwrite can only be used to165        # build a filesystem pool, but it will fail for now166            # if pool_type == "fs":167            #    option = '--overwrite'168            if build_acl:169                result = virsh.pool_build(pool_name, option, **acl_dargs)170            else:171                result = virsh.pool_build(pool_name, option,172                                          ignore_status=True)173            check_exit_status(result, build_error)174            if build_error:175                # Redo under negative case to keep case continue176                result = virsh.pool_build(pool_name, option,177                                          ignore_status=True)178                check_exit_status(result)179        # Check directory pool permission after build it180        if dir_mode_check:181            expect_mode = "0755"182            expect_mode_s = "drwxr-xr-x"183            xml_mode = PoolXML.new_from_dumpxml(pool_name).mode184            dir_mode = utils.system_output("ls -dl %s" % pool_target)185            if expect_mode_s in dir_mode:186                logging.debug("%s mode is: %s", pool_target, dir_mode)187            else:188                raise error.TestFail("Unexpect mode of %s: %s" % pool_target,189                                     expect_mode_s)190            if xml_mode == expect_mode:191                logging.debug("Target path mode in pool XML is: %s",192                              expect_mode)193            else:194                raise error.TestFail("Unexpect mode of %s in pool XML:"195                                     " %s" % pool_target, dir_mode)196        # Step (4)197        # Pool start198        if start_acl:199            result = virsh.pool_start(pool_name, **acl_dargs)200        else:201            result = virsh.pool_start(pool_name, ignore_status=True)202        check_exit_status(result, start_error)203        if start_error:204            # Redo under negative case to keep case continue205            result = virsh.pool_start(pool_name, ignore_status=True)206            check_exit_status(result)207        option = "--persistent --type %s" % pool_type208        check_pool_list(pool_name, option)209        # Step (5)210        # Pool destroy211        if destroy_acl:212            result = virsh.pool_destroy(pool_name, **acl_dargs)213        else:214            result = virsh.pool_destroy(pool_name)215        if result:216            if destroy_error:217                raise error.TestFail("Expect fail, but run successfully.")218        else:219            if not destroy_error:220                raise error.TestFail("Pool %s destroy failed, not expected."221                                     % pool_name)222            else:223                # Redo under negative case to keep case continue224                if virsh.pool_destroy(pool_name):225                    logging.debug("Pool %s destroyed.", pool_name)226                else:227                    raise error.TestFail("Destroy pool % failed." % pool_name)228        # Step (6)229        # Pool refresh for 'dir' type pool230        # Pool start231        result = virsh.pool_start(pool_name, ignore_status=True)232        check_exit_status(result)233        if pool_type == "dir":234            os.mknod(vol_path)235            if refresh_acl:236                result = virsh.pool_refresh(pool_name, **acl_dargs)237            else:238                result = virsh.pool_refresh(pool_name)239            check_exit_status(result, refresh_error)240        # Step (7)241        # Pool vol-list242        if vol_list_acl:243            result = virsh.vol_list(pool_name, **acl_dargs)244        else:245            result = virsh.vol_list(pool_name)246        check_exit_status(result, vol_list_error)247        # Step (8)248        # Pool delete for 'dir' type pool249        if virsh.pool_destroy(pool_name):250            logging.debug("Pool %s destroyed.", pool_name)251        else:252            raise error.TestFail("Destroy pool % failed." % pool_name)253        if pool_type == "dir":254            if os.path.exists(vol_path):255                os.remove(vol_path)256            if delete_acl:257                result = virsh.pool_delete(pool_name, **acl_dargs)258            else:259                result = virsh.pool_delete(pool_name, ignore_status=True)260            check_exit_status(result, delete_error)261            option = "--inactive --type %s" % pool_type262            check_pool_list(pool_name, option)263            if not delete_error:264                if os.path.exists(pool_target):265                    raise error.TestFail("The target path '%s' still exist." %266                                         pool_target)267        result = virsh.pool_undefine(pool_name, ignore_status=True)268        check_exit_status(result)269        check_pool_list(pool_name, "--all", True)270    finally:271        # Clean up272        if os.path.exists(pool_xml):273            os.remove(pool_xml)274        if not _pool.delete_pool(pool_name):275            logging.error("Can't delete pool: %s", pool_name)276        if cleanup_env[2]:277            cmd = "pvs |grep %s |awk '{print $1}'" % vg_name278            pv_name = utils.system_output(cmd)279            lv_utils.vg_remove(vg_name)280            utils.run("pvremove %s" % pv_name)281        if cleanup_env[1]:282            utils_test.libvirt.setup_or_cleanup_iscsi(False)283        if cleanup_env[0]:284            utils_test.libvirt.setup_or_cleanup_nfs(...

test_docker_settings.py

Source:test_docker_settings.py

2import pytest3from puppetboard import docker_settings4from importlib import reload as reload5@pytest.fixture(scope='function')6def cleanup_env(request):7    for env_var in dir(docker_settings):8        if (env_var.startswith('__') or env_var.startswith('_') or9                env_var.islower()):10            continue11        if env_var in os.environ:12            del os.environ[env_var]13    reload(docker_settings)14    return15def test_default_host_port(cleanup_env):16    assert docker_settings.PUPPETDB_HOST == 'puppetdb'17    assert docker_settings.PUPPETDB_PORT == 808018def test_set_host_port(cleanup_env):19    os.environ['PUPPETDB_HOST'] = 'puppetdb2'20    os.environ['PUPPETDB_PORT'] = '9081'...

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.