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

How to use poll_proc method in avocado

Best Python code snippet using avocado_python

test_dogwrap.py

Source:test_dogwrap.py

...81            parse_options(['--proc_poll_interval', 'invalid'])82        with mock.patch.dict(os.environ, values={"DD_API_KEY": "the_key"}, clear=True):83            options, _ = parse_options([])84            self.assertEqual(options.api_key, "the_key")85    def test_poll_proc(self):86        mock_proc = mock.Mock()87        mock_proc.poll.side_effect = [None, 0]88        return_value = poll_proc(mock_proc, 0.1, 1)89        self.assertEqual(return_value, 0)90        self.assertEqual(mock_proc.poll.call_count, 2)91    def test_poll_timeout(self):92        mock_proc = mock.Mock()93        mock_proc.poll.side_effect = [None, None, None]94        with self.assertRaises(Timeout):95            poll_proc(mock_proc, 0.1, 0.2)96    @mock.patch('datadog.dogshell.wrap.poll_proc')97    @mock.patch('subprocess.Popen')98    def test_execute(self, mock_popen, mock_poll):99        mock_proc = mock.Mock()100        mock_proc.stdout.readline.side_effect = [b'out1\n', b'']101        mock_proc.stderr.readline.side_effect = [b'err1\n', b'']102        mock_popen.return_value = mock_proc103        mock_poll.return_value = 0104        return_code, stdout, stderr, duration = execute('foo', 10, 20, 30, 1, False)105        self.assertEqual(return_code, 0)106        self.assertEqual(stdout, b'out1\n')107        self.assertEqual(stderr, b'err1\n')108        mock_popen.assert_called_once()109        mock_poll.assert_called_once_with(mock_proc, 1, 10)...

wrap.py

Source:wrap.py

...18from dogapi.common import get_ec2_instance_id19SUCCESS = 'success'20ERROR = 'error'21class Timeout(Exception): pass22def poll_proc(proc, sleep_interval, timeout):23    start_time = time.time()24    returncode = None25    while returncode is None:26        returncode = proc.poll()27        if time.time() - start_time > timeout:28            raise Timeout()29        else:30            time.sleep(sleep_interval)31    return returncode32def execute(cmd, cmd_timeout, sigterm_timeout, sigkill_timeout,33            proc_poll_interval):34    start_time = time.time()35    returncode = -136    stdout = ''37    stderr = ''38    try:39        proc = subprocess.Popen(u' '.join(cmd), stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)40    except Exception:41        print >> sys.stderr, u"Failed to execute %s" % (repr(cmd))42        raise43    try:44        returncode = poll_proc(proc, proc_poll_interval, cmd_timeout)45        stdout, stderr = proc.communicate()46        duration = time.time() - start_time47    except Timeout:48        duration = time.time() - start_time49        try:50            proc.terminate()51            sigterm_start = time.time()52            try:53                print >> sys.stderr, "Command timed out after %.2fs, killing with SIGTERM" % (time.time() - start_time)54                poll_proc(proc, proc_poll_interval, sigterm_timeout)55                returncode = Timeout56            except Timeout:57                print >> sys.stderr, "SIGTERM timeout failed after %.2fs, killing with SIGKILL" % (time.time() - sigterm_start)58                proc.kill()59                poll_proc(proc, proc_poll_interval, sigkill_timeout)60                returncode = Timeout61        except OSError as e:62            # Ignore OSError 3: no process found.63            if e.errno != 3:64                raise65    return returncode, stdout, stderr, duration66def main():67    parser = OptionParser()68    parser.add_option('-n', '--name', action='store', type='string', help="The name of the event")69    parser.add_option('-k', '--api_key', action='store', type='string')70    parser.add_option('-m', '--submit_mode', action='store', type='choice', default='errors', choices=['errors', 'all'])71    parser.add_option('-t', '--timeout', action='store', type='int', default=60*60*24)72    parser.add_option('--sigterm_timeout', action='store', type='int', default=60*2)73    parser.add_option('--sigkill_timeout', action='store', type='int', default=60)...

caejd.py

Source:caejd.py

1import logging.config2from multiprocessing import Process3import os4import sys5import time6import subprocess7import django.conf8# Adding the project directory to the path to make imports of other modules9# of the project possible.10TOP_LEVEL_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))11sys.path.insert(0, TOP_LEVEL_DIR)12from utils.graceful_killer import GracefulKiller  # noqa: E40213from utils.jobinfo import poll, update  # noqa: E40214from utils.logger_copy import copy_logger_settings  # noqa: E40215# -----------------------------------------------------------------------------16# Logging Setting17# -----------------------------------------------------------------------------18logging.config.dictConfig(django.conf.settings.LOGGING)19# -----------------------------------------------------------------------------20# Django Server21# -----------------------------------------------------------------------------22def run_django_server():23    graceful_killer = GracefulKiller(name="Django")24    manage_path = os.path.join(TOP_LEVEL_DIR, "manage.py")25    django_call = ["python", manage_path, "runserver", "0:8000"]26    django_proc = subprocess.Popen(27        django_call,28        stderr=subprocess.STDOUT,29        universal_newlines=True30    )31    while True:32        time.sleep(1)33        if graceful_killer.kill_now:34            django_proc.terminate()35            break36# -----------------------------------------------------------------------------37# RUN38# -----------------------------------------------------------------------------39if __name__ == "__main__":40    logger = logging.getLogger(__name__)41    copy_logger_settings(logger.name, "utils.graceful_killer")42    logger.info("CAEJobDiary started...")43    procs = []44    logger.info("Starting polling process.")45    poll_proc = Process(46        target=poll.main,47        name="Polling")48    poll_proc.start()49    procs.append(poll_proc)50    logger.info("Starting update process.")51    update_proc = Process(52        target=update.main,53        name="Update")54    update_proc.start()55    procs.append(update_proc)56    logger.info("Starting Django server.")57    django_server_proc = Process(58        target=run_django_server,59        name="Django Server")60    django_server_proc.start()61    procs.append(django_server_proc)62    try:63        while True:64            time.sleep(1)65    except KeyboardInterrupt:66        logger.info("Keyboard interruption detected. Stopping processes.")67        for proc in procs:68            logger.info("Stopping process: (pid {}) {}".format(69                proc.pid, proc.name))70            proc.join()71            proc.terminate()72            if proc.exitcode == 0:73                logger.info("Process {} ended successfully.".format(74                    proc.pid))75            else:76                logger.warning("Process {} exitcode: {}".format(77                    proc.pid, proc.exitcode))78    except Exception as err_msg:79        logger.exception(80            "Exception in main process occurred!\n{}".format(err_msg))81        raise82    logger.info("All CAEJobDiary processes stopped.")...

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.