Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use test_distributed method in locust

Best Python code snippet using locust

ct_run.py

Source:ct_run.py

1#!/usr/bin/env python2"""Author: Konrad Zemek3Copyright (C) 2015 ACK CYFRONET AGH4This software is released under the MIT license cited in 'LICENSE.txt'5Runs oneprovider integration tests, providing Erlang's ct_run with every6environmental argument it needs for successful run. The output is put into7'test_distributed/logs'. The (init|end)_per_suite "testcases" are removed from8the surefire.xml output.9All paths used are relative to script's path, not to the running user's CWD.10Run the script with -h flag to learn about script's running options.11"""12from __future__ import print_function13import argparse14import glob15import os16import platform17import sys18import time19import re20import shutil21import json22sys.path.insert(0, 'bamboos/docker')23from environment import docker24parser = argparse.ArgumentParser(25    formatter_class=argparse.ArgumentDefaultsHelpFormatter,26    description='Run Common Tests.')27parser.add_argument(28    '--image', '-i',29    action='store',30    default='onedata/worker:v25',31    help='docker image to use as a test master',32    dest='image')33parser.add_argument(34    '--suite', '-s',35    action='append',36    help='name of the test suite',37    dest='suites')38parser.add_argument(39    '--case', '-c',40    action='append',41    help='name of the test case',42    dest='cases')43parser.add_argument(44    '--performance', '-p',45    action='store_true',46    default=False,47    help='run performance tests',48    dest='performance')49parser.add_argument(50    '--cover',51    action='store_true',52    default=False,53    help='run cover analysis',54    dest='cover')55parser.add_argument(56    '--stress',57    action='store_true',58    default=False,59    help='run stress tests',60    dest='stress')61parser.add_argument(62    '--stress-no-clearing',63    action='store_true',64    default=False,65    help='run stress tests without clearing data between test cases',66    dest='stress_no_clearing')67parser.add_argument(68    '--stress-time',69    action='store',70    help='time of stress test in sek',71    dest='stress_time')72parser.add_argument(73    '--auto-compile',74    action='store_true',75    default=False,76    help='compile test suites before run',77    dest='auto_compile')78args = parser.parse_args()79script_dir = os.path.dirname(os.path.abspath(__file__))80uid = str(int(time.time()))81excl_mods = glob.glob(82    os.path.join(script_dir, 'test_distributed', '*.erl'))83excl_mods = [os.path.basename(item)[:-4] for item in excl_mods]84cover_template = os.path.join(script_dir, 'test_distributed', 'cover.spec')85new_cover = os.path.join(script_dir, 'test_distributed', 'cover_tmp.spec')86incl_dirs = []87with open(cover_template, 'r') as template, open(new_cover, 'w') as cover:88    for line in template:89        if 'incl_dirs_r' in line:90            dirs_string = re.search(r'\[(.*?)\]', line).group(1)91            incl_dirs = [os.path.join(script_dir, d[1:]) for d in92                         dirs_string.split(', ')]93            docker_dirs = [os.path.join('/root/build', d[1:-1]) for d in94                           dirs_string.split(', ')]95        elif 'excl_mods' in line:96            modules_string = re.search(r'\[(.*?)\]', line).group(1)97            excl_mods.extend([d.strip('"') for d in modules_string.split(', ')])98        else:99            print(line, file=cover)100    print('{{incl_dirs_r, ["{0}]}}.'.format(', "'.join(incl_dirs)), file=cover)101    print('{{excl_mods, [{0}]}}.'.format(102        ', '.join(excl_mods)), file=cover)103ct_command = ['ct_run',104              '-no_auto_compile' if not args.auto_compile else '',105              '-abort_if_missing_suites',106              '-dir', '.',107              '-logdir', './logs/',108              '-ct_hooks', 'cth_surefire', '[{path, "surefire.xml"}]',109              '-noshell',110              '-name', 'testmaster@testmaster.{0}.dev.docker'.format(uid),111              '-include', '../include', '../deps']112code_paths = ['-pa']113if incl_dirs:114    code_paths.extend([os.path.join(script_dir, item[:-1])115                       for item in incl_dirs])116else:117    code_paths.extend([os.path.join(script_dir, 'ebin')])118code_paths.extend(glob.glob(os.path.join(script_dir, 'deps', '*', 'ebin')))119ct_command.extend(code_paths)120if args.suites:121    ct_command.append('-suite')122    ct_command.extend(args.suites)123if args.cases:124    ct_command.append('-case')125    ct_command.extend(args.cases)126if args.stress_time:127    ct_command.extend(['-env', 'stress_time', args.stress_time])128if args.performance:129    ct_command.extend(['-env', 'performance', 'true'])130elif args.stress:131    ct_command.extend(['-env', 'stress', 'true'])132elif args.stress_no_clearing:133    ct_command.extend(['-env', 'stress_no_clearing', 'true'])134elif args.cover:135    ct_command.extend(['-cover', 'cover_tmp.spec'])136    env_descs = glob.glob(137        os.path.join(script_dir, 'test_distributed', '*', 'env_desc.json'))138    for file in env_descs:139        shutil.copyfile(file, file + '.bak')140        with open(file, 'r') as jsonFile:141            data = json.load(jsonFile)142            configs_to_change = []143            if 'provider_domains' in data:144                for provider in data['provider_domains']:145                    if 'op_worker' in data['provider_domains'][provider]:146                        configs_to_change.extend(147                            data['provider_domains'][provider][148                                'op_worker'].values())149                    if 'cluster_manager' in data['provider_domains'][provider]:150                        configs_to_change.extend(151                            data['provider_domains'][provider][152                                'cluster_manager'].values())153            if 'globalregistry_domains' in data:154                for globalregistry in data['globalregistry_domains']:155                    configs_to_change.extend(156                        data['globalregistry_domains'][globalregistry][157                            'globalregistry'].values())158            for config in configs_to_change:159                config['sys.config']['covered_dirs'] = docker_dirs160                config['sys.config']['covered_excluded_modules'] = excl_mods161            with open(file, 'w') as jsonFile:162                jsonFile.write(json.dumps(data))163command = '''164import os, subprocess, sys, stat165if {shed_privileges}:166    os.environ['HOME'] = '/tmp'167    docker_gid = os.stat('/var/run/docker.sock').st_gid168    os.chmod('/etc/resolv.conf', 0o666)169    os.setgroups([docker_gid])170    os.setregid({gid}, {gid})171    os.setreuid({uid}, {uid})172command = {cmd}173ret = subprocess.call(command)174import xml.etree.ElementTree as ElementTree, glob, re175for file in glob.glob('logs/*/surefire.xml'):176    tree = ElementTree.parse(file)177    for suite in tree.findall('.//testsuite'):178        for test in suite.findall('testcase'):179            match = re.match('(init|end)_per_suite', test.attrib['name'])180            if match is not None:181                suite.remove(test)182    tree.write(file)183sys.exit(ret)184'''185command = command.format(186    uid=os.geteuid(),187    gid=os.getegid(),188    cmd=ct_command,189    shed_privileges=(platform.system() == 'Linux'))190ret = docker.run(tty=True,191                 rm=True,192                 interactive=True,193                 workdir=os.path.join(script_dir, 'test_distributed'),194                 reflect=[(script_dir, 'rw'),195                          ('/var/run/docker.sock', 'rw')],196                 name='testmaster_{0}'.format(uid),197                 hostname='testmaster.{0}.dev.docker'.format(uid),198                 image=args.image,199                 command=['python', '-c', command])200os.remove(new_cover)201if args.cover:202    for file in env_descs:203        os.remove(file)204        shutil.move(file + '.bak', file)...

testing.py

Source:testing.py

1from dAuth.tests import test_database, test_distributed2def main():3    test_database.run_tests()4    # test_distributed.run_test()5    6if __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.