How to use process_is_alive method in autotest

Best Python code snippet using autotest_python

test_virsh.py

Source:test_virsh.py

...7# simple magic for using scripts within a source tree8basedir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))9if os.path.isdir(os.path.join(basedir, 'virttest')):10    sys.path.append(basedir)11def process_is_alive(name_pattern):12    """13    'pgrep name' misses all python processes and also long process names.14    'pgrep -f name' gets all shell commands with name in args.15    So look only for command whose initial pathname ends with name.16    Name itself is an egrep pattern, so it can use | etc for variations.17    """18    return process.system("pgrep -f '^([^ /]*/)*(%s)([ ]|$)'" % name_pattern,19                          ignore_status=True, shell=True) == 020class bogusVirshFailureException(unittest.TestCase.failureException):21    def __init__(self, *args, **dargs):22        self.virsh_args = args23        self.virsh_dargs = dargs24    def __str__(self):25        msg = ("Codepath under unittest attempted call to un-mocked virsh"26               " method, with args: '%s' and dargs: '%s'"27               % (self.virsh_args, self.virsh_dargs))28        return msg29def FakeVirshFactory(preserve=None):30    """31    Return Virsh() instance with methods to raise bogusVirshFailureException.32    Users of this class should override methods under test on instance.33    :param preserve: List of symbol names NOT to modify, None for all34    """35    from virttest import virsh36    def raise_bogusVirshFailureException(*args, **dargs):37        raise bogusVirshFailureException()38    if preserve is None:39        preserve = []40    fake_virsh = virsh.Virsh(virsh_exec='/bin/false',41                             uri='qemu:///system', debug=True,42                             ignore_status=True)43    # Make all virsh commands throw an exception by calling it44    for symbol in dir(virsh):45        # Get names of just closure functions by Virsh class46        if symbol in virsh.NOCLOSE + preserve:47            continue48        if isinstance(getattr(fake_virsh, symbol), virsh.VirshClosure):49            # fake_virsh is a propcan, can't use setattr.50            fake_virsh.__super_set__(symbol, raise_bogusVirshFailureException)51    return fake_virsh52class ModuleLoad(unittest.TestCase):53    from virttest import virsh54class ConstantsTest(ModuleLoad):55    def test_ModuleLoad(self):56        self.assertTrue(hasattr(self.virsh, 'NOCLOSE'))57        self.assertTrue(hasattr(self.virsh, 'SCREENSHOT_ERROR_COUNT'))58        self.assertTrue(hasattr(self.virsh, 'VIRSH_COMMAND_CACHE'))59        self.assertTrue(hasattr(self.virsh, 'VIRSH_EXEC'))60class TestVirshClosure(ModuleLoad):61    @staticmethod62    def somefunc(*args, **dargs):63        return (args, dargs)64    class SomeClass(dict):65        def somemethod(self):66            return "foobar"67    def test_init(self):68        # save some typing69        VC = self.virsh.VirshClosure70        # self is guaranteed to be not dict-like71        self.assertRaises(ValueError, VC, self.somefunc, self)72        self.assertRaises(ValueError, VC, lambda: None, self)73    def test_args(self):74        # save some typing75        VC = self.virsh.VirshClosure76        tcinst = self.SomeClass()77        vcinst = VC(self.somefunc, tcinst)78        args, dargs = vcinst('foo')79        self.assertEqual(len(args), 1)80        self.assertEqual(args[0], 'foo')81        self.assertEqual(len(dargs), 0)82    def test_fake_virsh(self):83        fake_virsh = FakeVirshFactory()84        for symb in dir(self.virsh):85            if symb in self.virsh.NOCLOSE:86                continue87            value = fake_virsh.__super_get__(symb)88            self.assertRaises(unittest.TestCase.failureException, value)89    def test_dargs(self):90        # save some typing91        VC = self.virsh.VirshClosure92        tcinst = self.SomeClass(foo='bar')93        vcinst = VC(self.somefunc, tcinst)94        args, dargs = vcinst()95        self.assertEqual(len(args), 0)96        self.assertEqual(len(dargs), 1)97        self.assertEqual(list(dargs.keys()), ['foo'])98        self.assertEqual(list(dargs.values()), ['bar'])99    def test_args_and_dargs(self):100        # save some typing101        VC = self.virsh.VirshClosure102        tcinst = self.SomeClass(foo='bar')103        vcinst = VC(self.somefunc, tcinst)104        args, dargs = vcinst('foo')105        self.assertEqual(len(args), 1)106        self.assertEqual(args[0], 'foo')107        self.assertEqual(len(dargs), 1)108        self.assertEqual(list(dargs.keys()), ['foo'])109        self.assertEqual(list(dargs.values()), ['bar'])110    def test_args_dargs_subclass(self):111        # save some typing112        VC = self.virsh.VirshClosure113        tcinst = self.SomeClass(foo='bar')114        vcinst = VC(self.somefunc, tcinst)115        args, dargs = vcinst('foo')116        self.assertEqual(len(args), 1)117        self.assertEqual(args[0], 'foo')118        self.assertEqual(len(dargs), 1)119        self.assertEqual(list(dargs.keys()), ['foo'])120        self.assertEqual(list(dargs.values()), ['bar'])121    def test_update_args_dargs_subclass(self):122        # save some typing123        VC = self.virsh.VirshClosure124        tcinst = self.SomeClass(foo='bar')125        vcinst = VC(self.somefunc, tcinst)126        args, dargs = vcinst('foo')127        self.assertEqual(len(args), 1)128        self.assertEqual(args[0], 'foo')129        self.assertEqual(len(dargs), 1)130        self.assertEqual(list(dargs.keys()), ['foo'])131        self.assertEqual(list(dargs.values()), ['bar'])132        # Update dictionary133        tcinst['sna'] = 'fu'134        # Is everything really the same?135        args, dargs = vcinst('foo', 'baz')136        self.assertEqual(len(args), 2)137        self.assertEqual(args[0], 'foo')138        self.assertEqual(args[1], 'baz')139        self.assertEqual(len(dargs), 2)140        self.assertEqual(dargs['foo'], 'bar')141        self.assertEqual(dargs['sna'], 'fu')142    def test_multi_inst(self):143        # save some typing144        VC1 = self.virsh.VirshClosure145        VC2 = self.virsh.VirshClosure146        tcinst1 = self.SomeClass(darg1=1)147        tcinst2 = self.SomeClass(darg1=2)148        vcinst1 = VC1(self.somefunc, tcinst1)149        vcinst2 = VC2(self.somefunc, tcinst2)150        args1, dargs1 = vcinst1(1)151        args2, dargs2 = vcinst2(2)152        self.assertEqual(len(args1), 1)153        self.assertEqual(len(args2), 1)154        self.assertEqual(args1[0], 1)155        self.assertEqual(args2[0], 2)156        self.assertEqual(len(dargs1), 1)157        self.assertEqual(len(dargs2), 1)158        self.assertEqual(dargs1['darg1'], 1)159        self.assertEqual(dargs2['darg1'], 2)160class ConstructorsTest(ModuleLoad):161    def test_VirshBase(self):162        vb = self.virsh.VirshBase()163        del vb  # keep pylint happy164    def test_Virsh(self):165        v = self.virsh.Virsh()166        del v  # keep pylint happy167    def test_VirshPersistent(self):168        test_virsh = self.virsh.Virsh()169        if test_virsh['virsh_exec'] == '/bin/true':170            return171        else:172            logging.disable(logging.INFO)173            vp = self.virsh.VirshPersistent()174            vp.close_session()  # Make sure session gets cleaned up175    def TestVirshClosure(self):176        class MyDict(dict):177            pass178        vc = self.virsh.VirshClosure(None, MyDict())179        del vc  # keep pylint happy180# Ensure the following tests ONLY run if a valid virsh command exists #####181class ModuleLoadCheckVirsh(unittest.TestCase):182    from virttest import virsh183    def run(self, *args, **dargs):184        test_virsh = self.virsh.Virsh()185        if test_virsh['virsh_exec'] == '/bin/true':186            return  # Don't run any tests, no virsh executable was found187        else:188            super(ModuleLoadCheckVirsh, self).run(*args, **dargs)189class SessionManagerTest(ModuleLoadCheckVirsh):190    def test_del_VirshPersistent(self):191        """192        Unittest for __del__ of VirshPersistent.193        This test makes sure the __del__ method of VirshPersistent works194        well in `del vp_instance`.195        """196        vp = self.virsh.VirshPersistent()197        virsh_exec = vp.virsh_exec198        self.assertTrue(process_is_alive(virsh_exec))199        del vp200        self.assertFalse(process_is_alive(virsh_exec))201    def test_VirshSession(self):202        """203        Unittest for VirshSession.204        This test use VirshSession over VirshPersistent with auto_close=True.205        """206        virsh_exec = self.virsh.Virsh()['virsh_exec']207        # Build a VirshSession object.208        session_1 = self.virsh.VirshSession(virsh_exec, auto_close=True)209        self.assertTrue(process_is_alive(virsh_exec))210        del session_1211        self.assertFalse(process_is_alive(virsh_exec))212    def test_VirshPersistent(self):213        """214        Unittest for session manager of VirshPersistent.215        """216        virsh_exec = self.virsh.Virsh()['virsh_exec']217        vp_1 = self.virsh.VirshPersistent()218        self.assertTrue(process_is_alive(virsh_exec))219        # Init the vp_2 with same params of vp_1.220        vp_2 = self.virsh.VirshPersistent(**vp_1)221        # Make sure vp_1 and vp_2 are refer to the same session.222        self.assertEqual(vp_1.session_id, vp_2.session_id)223        del vp_1224        # Make sure the session is not closed when vp_2 still refer to it.225        self.assertTrue(process_is_alive(virsh_exec))226        del vp_2227        # Session was closed since no other VirshPersistent refer to it.228        self.assertFalse(process_is_alive(virsh_exec))229class VirshHasHelpCommandTest(ModuleLoadCheckVirsh):230    def setUp(self):231        # subclasses override self.virsh232        self.VIRSH_COMMAND_CACHE = self.virsh.VIRSH_COMMAND_CACHE233    def test_false_command(self):234        self.assertFalse(self.virsh.has_help_command('print'))235        self.assertFalse(self.virsh.has_help_command('Commands:'))236        self.assertFalse(self.virsh.has_help_command('dom'))237        self.assertFalse(self.virsh.has_help_command('pool'))238    def test_true_command(self):239        self.assertTrue(self.virsh.has_help_command('uri'))240        self.assertTrue(self.virsh.has_help_command('help'))241        self.assertTrue(self.virsh.has_help_command('list'))242    def test_no_cache(self):243        self.VIRSH_COMMAND_CACHE = None244        self.assertTrue(self.virsh.has_help_command('uri'))245        self.VIRSH_COMMAND_CACHE = []246        self.assertTrue(self.virsh.has_help_command('uri'))247    def test_subcommand_help(self):248        regex = r'--command'249        self.assertTrue(self.virsh.has_command_help_match('help', regex))250        self.assertFalse(self.virsh.has_command_help_match('uri', regex))251    def test_groups_in_commands(self):252        # groups will be empty in older libvirt, but test will still work253        groups = self.virsh.help_command_group(cache=True)254        groups_set = set(groups)255        commands = self.virsh.help_command_only(cache=True)256        commands_set = set(commands)257        grp_cmd = self.virsh.help_command(cache=True)258        grp_cmd_set = set(grp_cmd)259        # No duplicates check260        self.assertEqual(len(commands_set), len(commands))261        self.assertEqual(len(groups_set), len(groups))262        self.assertEqual(len(grp_cmd_set), len(grp_cmd))263        # No groups in commands or commands in groups264        self.assertEqual(len(groups_set & commands_set), 0)265        # Groups and Commands in help_command266        self.assertTrue(len(grp_cmd_set), len(commands_set) + len(groups_set))267class VirshHelpCommandTest(ModuleLoadCheckVirsh):268    def test_cache_command(self):269        l1 = self.virsh.help_command(cache=True)270        l2 = self.virsh.help_command()271        l3 = self.virsh.help_command()272        self.assertEqual(l1, l2)273        self.assertEqual(l2, l3)274        self.assertEqual(l3, l1)275class VirshClassHasHelpCommandTest(VirshHasHelpCommandTest):276    def setUp(self):277        logging.disable(logging.INFO)278        super(VirshClassHasHelpCommandTest, self).setUp()279        self.virsh = self.virsh.Virsh(debug=False)280class VirshPersistentClassHasHelpCommandTest(VirshHasHelpCommandTest):281    def setUp(self):282        logging.disable(logging.INFO)283        super(VirshPersistentClassHasHelpCommandTest, self).setUp()284        self.VirshPersistent = self.virsh.VirshPersistent285        self.virsh = self.VirshPersistent(debug=False)286        self.assertTrue(process_is_alive(self.virsh.virsh_exec))287    def test_recycle_session(self):288        # virsh can be used as a dict of it's properties289        another = self.VirshPersistent(**self.virsh)290        self.assertEqual(self.virsh.session_id, another.session_id)291    def tearDown(self):292        self.assertTrue(process_is_alive(self.virsh.virsh_exec))293        self.virsh.close_session()294        self.assertFalse(process_is_alive(self.virsh.virsh_exec))295if __name__ == '__main__':...

virsh_unittest.py

Source:virsh_unittest.py

...139        well in `del vp_instance`.140        """141        vp = self.virsh.VirshPersistent()142        virsh_exec = vp.virsh_exec143        self.assertTrue(utils.process_is_alive(virsh_exec))144        del vp145        self.assertFalse(utils.process_is_alive(virsh_exec))146    def test_VirshSession(self):147        """148        Unittest for VirshSession.149        This test use VirshSession over VirshPersistent with auto_close=True.150        """151        virsh_exec = self.virsh.Virsh()['virsh_exec']152        # Build a VirshSession object.153        session_1 = self.virsh.VirshSession(virsh_exec, auto_close=True)154        self.assertTrue(utils.process_is_alive(virsh_exec))155        del session_1156        self.assertFalse(utils.process_is_alive(virsh_exec))157    def test_VirshPersistent(self):158        """159        Unittest for session manager of VirshPersistent.160        """161        virsh_exec = self.virsh.Virsh()['virsh_exec']162        vp_1 = self.virsh.VirshPersistent()163        self.assertTrue(utils.process_is_alive(virsh_exec))164        # Init the vp_2 with same params of vp_1.165        vp_2 = self.virsh.VirshPersistent(**vp_1)166        # Make sure vp_1 and vp_2 are refer to the same session.167        self.assertEqual(vp_1.session_id, vp_2.session_id)168        del vp_1169        # Make sure the session is not closed when vp_2 still refer to it.170        self.assertTrue(utils.process_is_alive(virsh_exec))171        del vp_2172        # Session was closed since no other VirshPersistent refer to it.173        self.assertFalse(utils.process_is_alive(virsh_exec))174class VirshHasHelpCommandTest(ModuleLoadCheckVirsh):175    def setUp(self):176        # subclasses override self.virsh177        self.VIRSH_COMMAND_CACHE = self.virsh.VIRSH_COMMAND_CACHE178    def test_false_command(self):179        self.assertFalse(self.virsh.has_help_command('print'))180        self.assertFalse(self.virsh.has_help_command('Commands:'))181        self.assertFalse(self.virsh.has_help_command('dom'))182        self.assertFalse(self.virsh.has_help_command('pool'))183    def test_true_command(self):184        self.assertTrue(self.virsh.has_help_command('uri'))185        self.assertTrue(self.virsh.has_help_command('help'))186        self.assertTrue(self.virsh.has_help_command('list'))187    def test_no_cache(self):188        self.VIRSH_COMMAND_CACHE = None189        self.assertTrue(self.virsh.has_help_command('uri'))190        self.VIRSH_COMMAND_CACHE = []191        self.assertTrue(self.virsh.has_help_command('uri'))192    def test_subcommand_help(self):193        regex = r'\s+\[--command\]\s+\<string\>\s+'194        self.assertTrue(self.virsh.has_command_help_match('help', regex))195        self.assertFalse(self.virsh.has_command_help_match('uri', regex))196    def test_groups_in_commands(self):197        # groups will be empty in older libvirt, but test will still work198        groups = self.virsh.help_command_group(cache=True)199        groups_set = set(groups)200        commands = self.virsh.help_command_only(cache=True)201        commands_set = set(commands)202        grp_cmd = self.virsh.help_command(cache=True)203        grp_cmd_set = set(grp_cmd)204        # No duplicates check205        self.assertEqual(len(commands_set), len(commands))206        self.assertEqual(len(groups_set), len(groups))207        self.assertEqual(len(grp_cmd_set), len(grp_cmd))208        # No groups in commands or commands in groups209        self.assertEqual(len(groups_set & commands_set), 0)210        # Groups and Commands in help_command211        self.assertTrue(len(grp_cmd_set), len(commands_set) + len(groups_set))212class VirshHelpCommandTest(ModuleLoadCheckVirsh):213    def test_cache_command(self):214        l1 = self.virsh.help_command(cache=True)215        l2 = self.virsh.help_command()216        l3 = self.virsh.help_command()217        self.assertEqual(l1, l2)218        self.assertEqual(l2, l3)219        self.assertEqual(l3, l1)220class VirshClassHasHelpCommandTest(VirshHasHelpCommandTest):221    def setUp(self):222        logging.disable(logging.INFO)223        super(VirshClassHasHelpCommandTest, self).setUp()224        self.virsh = self.virsh.Virsh(debug=False)225class VirshPersistentClassHasHelpCommandTest(VirshHasHelpCommandTest):226    def setUp(self):227        logging.disable(logging.INFO)228        super(VirshPersistentClassHasHelpCommandTest, self).setUp()229        self.VirshPersistent = self.virsh.VirshPersistent230        self.virsh = self.VirshPersistent(debug=False)231        self.assertTrue(utils.process_is_alive(self.virsh.virsh_exec))232    def test_recycle_session(self):233        # virsh can be used as a dict of it's properties234        another = self.VirshPersistent(**self.virsh)235        self.assertEqual(self.virsh.session_id, another.session_id)236    def tearDown(self):237        self.assertTrue(utils.process_is_alive(self.virsh.virsh_exec))238        self.virsh.close_session()239        self.assertFalse(utils.process_is_alive(self.virsh.virsh_exec))240if __name__ == '__main__':...

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.