Best Python code snippet using avocado_python
test_interrupt.py
Source:test_interrupt.py  
...76        return len(test_processes) == 077    def setUp(self):78        super(InterruptTest, self).setUp()79        self.test_module = None80    @skipOnLevelsInferiorThan(2)81    def test_badly_behaved_sigint(self):82        """83        Make sure avocado can cleanly get out of a loop of badly behaved tests.84        :avocado: tags=parallel:185        """86        bad_test_basename = ('wontquit-%s' %87                             data_factory.generate_random_string(5))88        bad_test = script.TemporaryScript(bad_test_basename, BAD_TEST,89                                          'avocado_interrupt_test',90                                          mode=DEFAULT_MODE)91        bad_test.save()92        self.test_module = bad_test.path93        os.chdir(BASEDIR)94        cmd = ('%s run %s --disable-sysinfo --job-results-dir %s ' %95               (AVOCADO, self.test_module, self.tmpdir.name))96        proc = subprocess.Popen(cmd.split(),97                                stdout=subprocess.PIPE,98                                stderr=subprocess.STDOUT)99        if not wait.wait_for(lambda: self.has_children(proc), timeout=10):100            process.kill_process_tree(proc.pid)101            self.fail('Avocado did not start the test process.')102        # This test will ignore SIGINT, so it should terminate103        # when we send the second SIGINT.104        os.kill(proc.pid, signal.SIGINT)105        # We have to actually wait 2+ seconds until106        # the ignore window is over107        time.sleep(2.5)108        os.kill(proc.pid, signal.SIGINT)109        if not wait.wait_for(lambda: self.is_finished(proc), timeout=30):110            process.kill_process_tree(proc.pid)111            self.fail('Avocado was still running after receiving SIGINT '112                      'twice.')113        self.assertTrue(wait.wait_for(self._no_test_in_process_table,114                        timeout=10), 'Avocado left processes behind.')115        output = proc.stdout.read()116        # Make sure the Interrupted requested sentence is there117        self.assertIn(b'Interrupt requested. Waiting 2 seconds for test to '118                      b'finish (ignoring new Ctrl+C until then)', output)119        # Make sure the Killing test subprocess message did appear120        self.assertIn(b'Killing test subprocess', output)121    @skipOnLevelsInferiorThan(2)122    def test_badly_behaved_sigterm(self):123        """124        Make sure avocado can cleanly get out of a loop of badly behaved tests.125        :avocado: tags=parallel:1126        """127        bad_test_basename = ('wontquit-%s' %128                             data_factory.generate_random_string(5))129        bad_test = script.TemporaryScript(bad_test_basename, BAD_TEST,130                                          'avocado_interrupt_test',131                                          mode=DEFAULT_MODE)132        bad_test.save()133        self.test_module = bad_test.path134        os.chdir(BASEDIR)135        cmd = ('%s run %s --disable-sysinfo --job-results-dir %s ' %136               (AVOCADO, self.test_module, self.tmpdir.name))137        proc = subprocess.Popen(cmd.split(),138                                stdout=subprocess.PIPE,139                                stderr=subprocess.STDOUT)140        if not wait.wait_for(lambda: self.has_children(proc), timeout=10):141            process.kill_process_tree(proc.pid)142            self.fail('Avocado did not start the test process.')143        # This test should be terminated when the main process144        # receives a SIGTERM, even if the test process ignores SIGTERM.145        os.kill(proc.pid, signal.SIGTERM)146        if not wait.wait_for(lambda: self.is_finished(proc), timeout=10):147            process.kill_process_tree(proc.pid)148            self.fail('Avocado was still running after receiving SIGINT '149                      'twice.')150        self.assertTrue(wait.wait_for(self._no_test_in_process_table,151                        timeout=10), 'Avocado left processes behind.')152        # Make sure the Interrupted test sentence is there153        self.assertIn(b'Terminated\n', proc.stdout.read())154    @skipOnLevelsInferiorThan(2)155    def test_well_behaved_sigint(self):156        """157        Make sure avocado can cleanly get out of a loop of well behaved tests.158        :avocado: tags=parallel:1159        """160        good_test_basename = ('goodtest-%s.py' %161                              data_factory.generate_random_string(5))162        good_test = script.TemporaryScript(good_test_basename, GOOD_TEST,163                                           'avocado_interrupt_test',164                                           mode=DEFAULT_MODE)165        good_test.save()166        self.test_module = good_test.path167        os.chdir(BASEDIR)168        cmd = ('%s run %s --disable-sysinfo --job-results-dir %s ' %169               (AVOCADO, self.test_module, self.tmpdir.name))170        proc = subprocess.Popen(cmd.split(),171                                stdout=subprocess.PIPE,172                                stderr=subprocess.STDOUT)173        if not wait.wait_for(lambda: self.has_children(proc), timeout=10):174            process.kill_process_tree(proc.pid)175            self.fail('Avocado did not start the test process.')176        # This test will not ignore SIGINT, so it should177        # terminate right away.178        os.kill(proc.pid, signal.SIGINT)179        if not wait.wait_for(lambda: self.is_finished(proc), timeout=10):180            process.kill_process_tree(proc.pid)181            self.fail('Avocado was still running after receiving SIGINT '182                      'twice.')183        self.assertTrue(wait.wait_for(self._no_test_in_process_table,184                        timeout=10), 'Avocado left processes behind.')185        output = proc.stdout.read()186        # Make sure the Interrupted requested sentence is there187        self.assertIn(b'Interrupt requested. Waiting 2 seconds for test to '188                      b'finish (ignoring new Ctrl+C until then)', output)189        # Make sure the Killing test subprocess message is not there190        self.assertNotIn(b'Killing test subprocess', output)191    @skipOnLevelsInferiorThan(2)192    def test_well_behaved_sigterm(self):193        """194        Make sure avocado can cleanly get out of a loop of well behaved tests.195        :avocado: tags=parallel:1196        """197        good_test_basename = ('goodtest-%s.py' %198                              data_factory.generate_random_string(5))199        good_test = script.TemporaryScript(good_test_basename, GOOD_TEST,200                                           'avocado_interrupt_test',201                                           mode=DEFAULT_MODE)202        good_test.save()203        self.test_module = good_test.path204        os.chdir(BASEDIR)205        cmd = ('%s run %s --disable-sysinfo --job-results-dir %s ' %...test_utils.py
Source:test_utils.py  
...104        self.fake_uptime = os.path.join(self.tmpdir.name, 'uptime')105        with open(self.fake_uptime, 'w') as fake_uptime_obj:106            fake_uptime_obj.write(FAKE_UPTIME_CONTENTS)107        os.chmod(self.fake_uptime, DEFAULT_MODE)108    @skipOnLevelsInferiorThan(2)109    def test_process_start(self):110        """111        :avocado: tags=parallel:1112        """113        proc = process.SubProcess('%s 1 0' % self.fake_vmstat)114        proc.start()115        time.sleep(3)116        proc.terminate()117        proc.wait(timeout=1)118        stdout = proc.get_stdout().decode()119        self.assertIn('memory', stdout, 'result: %s' % stdout)120        self.assertRegex(stdout, '[0-9]+')121    @skipOnLevelsInferiorThan(2)122    def test_process_stop_interrupted(self):123        """124        :avocado: tags=parallel:1125        """126        proc = process.SubProcess('%s 1 3' % self.fake_vmstat)127        proc.start()128        time.sleep(3)129        proc.stop(2)130        result = proc.result131        self.assertIn('timeout after', result.interrupted, "Process wasn't interrupted")132    @skipOnLevelsInferiorThan(2)133    def test_process_stop_uninterrupted(self):134        """135        :avocado: tags=parallel:1136        """137        proc = process.SubProcess('%s 1 3' % self.fake_vmstat)138        proc.start()139        time.sleep(3)140        proc.stop(4)141        result = proc.result142        self.assertFalse(result.interrupted, "Process was interrupted to early")143    def test_process_run(self):144        proc = process.SubProcess(self.fake_uptime)145        result = proc.run()146        self.assertEqual(result.exit_status, 0, 'result: %s' % result)147        self.assertIn(b'load average', result.stdout)148def file_lock_action(args):149    path, players, max_individual_timeout = args150    max_timeout = max_individual_timeout * players151    with FileLock(path, max_timeout):152        sleeptime = random.random() / 100153        time.sleep(sleeptime)154class FileLockTest(TestCaseTmpDir):155    @skipOnLevelsInferiorThan(3)156    def test_filelock(self):157        """158        :avocado: tags=parallel:1159        """160        # Calculate the timeout based on t_100_iter + 2e-5*players161        start = time.time()162        for _ in range(100):163            with FileLock(self.tmpdir.name):164                pass165        timeout = 0.02 + (time.time() - start)166        players = 1000167        pool = multiprocessing.Pool(players)168        args = [(self.tmpdir.name, players, timeout)] * players169        try:...test_sysinfo.py
Source:test_sysinfo.py  
...110                path = tmp111            raise AssertionError("Sleep output not recorded in '%s', first "112                                 "existing location '%s' contains:\n%s"113                                 % (sleep_log, path, os.listdir(path)))114    @skipOnLevelsInferiorThan(2)115    def test_sysinfo_interrupted(self):116        """117        :avocado: tags=parallel:1118        """119        self.run_sysinfo_interrupted(10, 1, 15)120    @skipOnLevelsInferiorThan(2)121    def test_sysinfo_not_interrupted(self):122        """123        :avocado: tags=parallel:1124        """125        self.run_sysinfo_interrupted(5, -1, 10)126if __name__ == '__main__':...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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!
