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

How to use run_bg method in autotest

Best Python code snippet using autotest_python

update_plone

Source:update_plone

...14    zero_downtime = is_zero_downtime_configured()15    print 'UPDATE PLONE'16    print '${0} -> ${1}'.format(oldrev, newrev)17    CHANGED_FILES.extend(18        run_bg('git diff {0} {1} --name-only'.format(oldrev, newrev))19        .strip().splitlines())20    update_sources()21    print ''22    print ''23    buildout_required = has_changed(r'^.*\.cfg$') or has_changed(r'setup.py')24    print 'buildout required:', bool(buildout_required)25    print 'zero downtime:', zero_downtime26    sys.stdout.flush()27    maybe_start('maintenance')28    maybe_stop('instance0')29    maybe_stop('instancepub')30    if buildout_required:31        run_buildout()32        run_fg('bin/supervisorctl reread')33    maybe_restart('solr')34    maybe_restart('tika-server')35    maybe_restart('redis')36    maybe_start('instance0')37    proposed_upgrades = has_proposed_upgrades()38    plone_upgrades = plone_upgrade_needed()39    if proposed_upgrades or plone_upgrades:40        if not zero_downtime:41            stop_load_balanced_instances()42        if plone_upgrades:43            run_plone_upgrades()44        if proposed_upgrades:45            run_upgrades()46        restart_load_balanced_instances()47    else:48        restart_load_balanced_instances()49    update_supervisor_config()50    recook_resources()51    maybe_start('instancepub')52    maybe_stop('instance0')53def run_bg(cmd, cwd=None):54    if isinstance(cmd, unicode):55       cmd = cmd.encode('ascii','ignore')56    proc = subprocess.Popen(shlex.split(cmd),57                            stdout=subprocess.PIPE,58                            stderr=subprocess.PIPE,59                            cwd=cwd)60    stdout, stderr = proc.communicate()61    if proc.poll():62        print 'ERROR {0}'.format(cmd)63        print stdout64        print stderr65        sys.stdout.flush()66        sys.exit(1)67    return stdout68def run_fg(cmd, abort_on_error=True):69    LOG.info('> {0}'.format(cmd))70    print ''71    print '>', cmd72    sys.stdout.flush()73    if os.system(cmd):74        if abort_on_error:75            sys.exit(1)76        else:77            return False78    return True79def has_changed(file_regex):80    return filter(re.compile(file_regex).match, CHANGED_FILES)81def update_sources():82    for path in filter(os.path.isdir, map('src/'.__add__, os.listdir('src'))):83        if not os.path.isdir(os.path.join(path, '.git')):84            print ''85            print 'WARNING: not a GIT checkout:', path86            continue87        oldrev = run_bg('git rev-parse HEAD', cwd=path).strip()88        if not run_fg('(cd {0} && git pull)'.format(path), abort_on_error=False):89            continue90        newrev = run_bg('git rev-parse HEAD', cwd=path).strip()91        CHANGED_FILES.extend(92            map((path + os.sep).__add__,93                run_bg('git diff {0} {1} --name-only'.format(oldrev, newrev),94                       cwd=path).strip().splitlines()))95        print path, oldrev, '=>', newrev96        LOG.info('pulling {0}: {1} => {2}'.format(path, oldrev, newrev))97def assure_supervisord_running():98    try:99        subprocess.check_call(shlex.split('bin/supervisorctl avail'),100                              stdout=subprocess.PIPE,101                              stderr=subprocess.PIPE)102    except subprocess.CalledProcessError, exc:103        if exc.returncode != 2:104            raise105        run_fg('bin/supervisord')106def supervisor_status():107    assure_supervisord_running()108    return dict(map(lambda line: line.split()[:2],109                    run_bg('bin/supervisorctl status').strip().splitlines()))110def load_balanced_instances():111    whitelist = (112        'instance1', 'instance2', 'instance3', 'instance4', 'instance5',113        'instance6', 'instance7', 'instance8', 'instance9')114    for name, status in supervisor_status().items():115        if name in whitelist:116            yield name, status117def stop_load_balanced_instances():118    names = [name for (name, status) in load_balanced_instances()119             if status != 'STOPPED' and name != 'instance0']120    run_fg('bin/supervisorctl stop {0}'.format(' '.join(names)))121def restart_load_balanced_instances():122    for instance_name, status in load_balanced_instances():123        if instance_name == 'instance0':124            continue125        if status != 'STOPPED':126            run_fg('bin/supervisorctl stop {0}'.format(instance_name))127        run_fg('bin/supervisorctl update {0}'.format(instance_name))128        run_fg('bin/supervisorctl start {0}'.format(instance_name))129def maybe_stop(name):130    status = supervisor_status()131    if name in status and status[name] != 'STOPPED':132        run_fg('bin/supervisorctl stop {0}'.format(name))133def maybe_start(name):134    status = supervisor_status()135    if name in status and status[name] != 'RUNNING':136        run_fg('bin/supervisorctl update {0}'.format(name))137        run_fg('bin/supervisorctl start {0}'.format(name))138def maybe_restart(name):139    maybe_stop(name)140    maybe_start(name)141def run_buildout():142    run_fg('bin/buildout')143def run_upgrades():144    for site in get_sites():145        run_fg('bin/upgrade install -s %s --proposed' % site['path'])146    return True147def has_proposed_upgrades():148    for site in get_sites():149        command = './bin/upgrade list -s %s --upgrades --json' % site['path']150        data = json.loads(run_bg(command))151        if len(data) > 0:152            print 'Proposed upgrades found.'153            sys.stdout.flush()154            return True155    print 'Proposed upgrades: 0'156    sys.stdout.flush()157    return False158def plone_upgrade_needed():159    for site in get_sites():160        command = './bin/upgrade plone_upgrade_needed -s %s' % site['path']161        if json.loads(run_bg(command)):162            print 'Plone upgrade needed.'163            sys.stdout.flush()164            return True165    return False166def run_plone_upgrades():167    print 'Running Plone upgrades:'168    for site in get_sites():169        run_fg('bin/upgrade plone_upgrade -s %s' % site['path'])170    return True171def recook_resources():172    for site in get_sites():173        run_fg('bin/upgrade recook -s %s' % site['path'])174def update_supervisor_config():175    """176    "./bin/supervisorctl reread" has output when the supervisor configuration177    has changed, e.g.:178    $ ./bin/supervisorctl reread179    instance1: changed180    redis: available181    For updating the supervisor daemon with the new settings, use the "update"182    action:183    $ ./bin/supervisorctl update instance1184    instance1: stopped185    instance1: updated process group186    As this does restart the programs, we do not want to do it for all programs187    (not for zeo, for example) and we do not want to stop all programs at once.188    We also do it at the end of the update as we might already have updated189    programs earlier when restarting them.190    """191    blacklist = (192        # Do not restart "zeo" automatically as it may result in193        # an unexpected downtime.194        'zeo',195    )196    print ''197    print ''198    print ''199    updates = run_bg('bin/supervisorctl reread')200    print 'Supervisor configuration reread: ', updates201    LOG.info('Supervisor configuration reread: {0}'.format(updates))202    if ':' not in updates:203        # No config updates to processes204        return205    for line in updates.strip().splitlines():206        program, state = line.strip().split(': ', 1)207        if program in blacklist:208            msg = ('Changes on program "{0}" are not automatically updated'209                   ' because the program is blacklisted.'210                   ' Manual action is necessary!').format(program)211            LOG.warning(msg)212            print '-' * 30213            print 'WARNING:', msg214            print '-' * 30215            continue216        print 'Updating {0} ({1})'.format(program, state)217        run_fg('bin/supervisorctl update {0}'.format(program))218def get_sites():219    if len(SITES) == 0:220        SITES.extend(json.loads(run_bg('./bin/upgrade sites --json')))221        if len(SITES) == 0:222            print "ERROR: No site found."223            sys.exit(1)224        print 'Sites found:', len(SITES)225    return SITES226def setup_logging():227    logger = logging.getLogger()228    logger.setLevel(logging.INFO)229    handler = logging.FileHandler('var/log/deploy-plone.log')230    handler.setFormatter(logging.Formatter(231        '%(asctime)s %(levelname)s %(name)s (%(process)s): %(message)s'))232    logger.addHandler(handler)233def is_zero_downtime_configured():234    # The zero downtime configuration must be in the buildout.cfg (may be a...

test_plugin.py

Source:test_plugin.py

...10    assert process.get_command() == "true"11def test_process_kill(process):12    process.set_command("read")13    with pytest.raises(SystemError):14        process.run_bg()15    assert process.kill() is None16def test_echo_hello(process):17    process.set_command(["echo", "hello", "world"])18    process.run()19    time.sleep(WAITSTATUS)20    assert process.get_stdout() == "hello world"21    assert process.get_returncode() == 022def test_run_background_echo_hello(process):23    process.set_command(["/usr/bin/sh", "-c", "/usr/bin/sleep 0.1"])24    process.run_bg()25    time.sleep(WAITSTATUS * 2)26    assert process.get_status() == 027def test_run_background_echo_hello_fail(process):28    process.set_command(["/usr/bin/sh", "-c", "/usr/bin/sleep 0.1 ; false"])29    process.run_bg()30    time.sleep(WAITSTATUS * 2)31    assert process.get_status() == 132def test_run_background_status_poll_fails(process):33    process.set_command(["/usr/bin/sleep", "3"])34    process.run_bg()35    assert process.get_status() == "Running"36def test_run_background_status_poll(process):37    process.set_command(["/usr/bin/sleep", "0.1"])38    process.run_bg()39    time.sleep(WAITSTATUS * 2)40    assert process.get_status(poll=1) == 041# Factory tests42def test_proc_factory_int(process_factory):43    # TODO With mypy in place this should not pass44    process_factory(88)45def test_proc_factory(process_factory):46    proc1 = process_factory(["/usr/bin/sleep", "100"])47    proc2 = process_factory(["/usr/bin/sleep", "101"])48    proc1.run_bg()49    proc2.run_bg()50    time.sleep(0.1)51    proc1.kill()52    proc2.kill()53def test_proc_factory_with_prefix(process_factory):54    proc1 = process_factory(["/usr/bin/sleep", "100"], name="hundred")55    proc2 = process_factory(["/usr/bin/sleep", "101"])56    proc1.run_bg()57    proc2.run_bg()58def test_proc_factory_autokill(process_factory):59    proc1 = process_factory(["/usr/bin/sleep", "100"])60    proc2 = process_factory(["/usr/bin/sleep", "101"])61    proc1.run_bg()62    proc2.run_bg()63def test_proc_factory_one_not_called(process_factory):64    # Test will pass and we will see a warning in pytest65    # TODO: can we run an assertion on that?66    proc1 = process_factory(["/usr/bin/sleep", "100"])67    process_factory(["/usr/bin/sleep", "101"])68    proc1.run_bg()69def test_proc_factory_has_exited_with_error(process_factory):70    # args = ["/usr/bin/sh", "-c", "/usr/bin/sleep 0.1 ; false"]71    args = ["/usr/bin/false"]72    proc1 = process_factory(args)73    proc1.run_bg()74    time.sleep(WAITSTATUS)75    assert proc1.get_status(5) == 176    assert proc1.get_returncode() == 177    time.sleep(1)78def test_proc_factory_was_never_started(process_factory):79    # Will happen when we call without the full path or invalid arguents.80    # E.g. if os.fork failed for that process.81    # In this case we will see the last complaints in stderr of the process82    # since we fist fork and then exeve83    args = ["idonotexist"]84    proc1 = process_factory(args)85    with pytest.raises(SystemError):86        proc1.run_bg()87    assert proc1.get_status(5) is None88    with pytest.raises(SystemError):89        proc1.get_returncode()90def test_grep_stdout_fg(process):91    process.set_command(["echo", "hello", "world"])92    process.run()93    assert process.get_stdout() == "hello world"94    assert process.get_returncode() == 095def test_grep_stdout_bg(process):96    process.set_command(["/usr/bin/echo", "hello", "worldpeace"])97    process.run_bg()98    # time.sleep(0.1)99    time.sleep(WAITSTATUS)100    assert process.get_status() == 0101    assert process.background is True102    assert process.get_stdout() == "hello worldpeace"103    assert process.get_returncode() == 0104def test_grep_stderr_bg(process):105    # process.set_command(["/bin/sh", "-c", "echo not found", ">&2"])106    process.set_command(["/usr/bin/ls", "notthere"])107    process.run_bg()108    # time.sleep(0.1)109    time.sleep(WAITSTATUS)110    assert process.get_status() == 2111    assert process.background is True112    assert (113        process.get_stderr()114        == "/usr/bin/ls: cannot access 'notthere': No such file or directory"115    )...

test_scopes.py

Source:test_scopes.py

...5    logging.info(6        "That one runs without the module fixture (no module fixture requested yet)"7    )8    proc = process_factory("/usr/bin/true")9    proc.run_bg()10def test_proc_factory_module(process_factory_module):11    logging.info("here we create the fixture")12    proc = process_factory_module("/usr/bin/true")13    proc.run_bg()14def test_proc_factory_module_2(process_factory_module):15    logging.info("here we create the fixture 2")16    proc = process_factory_module("/usr/bin/true")17    proc.run_bg()18def test_proc_factory_function(process_factory):19    logging.info("here we run a process")20    proc = process_factory("/usr/bin/true")21    proc.run_bg()22def test_proc_factory_function_2(process_factory):23    logging.info("here we run a process 2")24    proc = process_factory("/usr/bin/true")25    proc.run_bg()26def dis_test_proc_factory_session(process_factory_session):27    """TODO Disabled: stdout of pytest appears in the logged stdout ???28    Do we have a os.fork going wild?29    """30    logging.info("here we run a session fixture")31    proc = process_factory_session("/usr/bin/true")32    proc.run_bg()33def dis_test_proc_factory_function_2(process_factory_session):34    logging.info("here we run a session fixture 2")35    proc = process_factory_session("/usr/bin/true")...

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.