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

How to use create_test_queue method in localstack

Best Python code snippet using localstack_python

machine.py

Source:machine.py

1"""Factory function and class for creating and storing machine-related information.2"""3import logging4from collections import namedtuple5from CIME.utils import get_project  # pylint: disable=import-error6from ctsm.joblauncher.job_launcher_factory import (7    create_job_launcher,8    JOB_LAUNCHER_NOBATCH,9)10# Value of create_test_queue for which we don't actually add a '--queue' option to11# create_test, but instead leave that value unspecified, allowing CIME to pick an12# appropriate queue for each test using its standard mechanisms.13CREATE_TEST_QUEUE_UNSPECIFIED = "unspecified"14logger = logging.getLogger(__name__)15# TODO(wjs, 2018-08-31) Turn this into a real class, with getter methods.16#17# An attempt to call get_scratch_dir() should raise a RuntimeError if self._scratch_dir is18# None. (That is, we allow a machine to be created with a scratch_dir of None - because19# for some applications we don't need a scratch_dir - but we raise an exception if the20# application tries to use the scratch_dir when it wasn't properly set, in order to print21# a meaningful error message rather than whatever message you'd get when trying to use the22# None value somewhere.)23#24# For now, just keep in mind that it's possible that a machine's scratch_dir may be None,25# if it wasn't set explicitly and there was no default available. For now, it's up to the26# user of the machine object to check for that possibility if need be.27#28# Similar notes apply to baseline_dir.29Machine = namedtuple(30    "Machine",31    [32        "name",  # str33        "scratch_dir",  # str34        "baseline_dir",  # str35        "account",  # str or None36        "create_test_retry",  # int37        "create_test_queue",  # str38        "job_launcher",39    ],40)  # subclass of JobLauncherBase41def create_machine(42    machine_name,43    defaults,44    job_launcher_type=None,45    scratch_dir=None,46    account=None,47    job_launcher_queue=None,48    job_launcher_walltime=None,49    job_launcher_nice_level=None,50    job_launcher_extra_args=None,51    allow_missing_entries=False,52):53    """Create a machine object (of type Machine, as given above)54    This uses the provided (non-None) arguments to override any defaults provided via the55    'defaults' argument.56    Args:57    machine_name (str): name of machine; this is used to index into the 'defaults'58        argument, and is used as the machine name in the returned object59    defaults: dict of MachineDefaults (as defined in machine_defaults)60    scratch_dir: path to scratch directory (if not provided, will attempt to get it from61        machine defaults)62    account: account to use for job submission to a queue (if not provided, will attempt63        to determine it automatically using the CIME method)64    job_launcher_type: one of the JOB_LAUNCHER constants defined in job_launcher_factory,65        or None. If None, we pick the default for this machine.66    job_launcher_queue (str or None): Queue to use (not applicable for67        JOB_LAUNCHER_NOBATCH and JOB_LAUNCHER_FAKE)68    job_launcher_walltime (str or None): Walltime to use (not applicable for69        JOB_LAUNCHER_NOBATCH and JOB_LAUNCHER_FAKE)70    job_launcher_nice_level (int or None): Level used for the nice command; only71        applicable for JOB_LAUNCHER_NOBATCH72    job_launcher_extra_args (str or None): Arguments to the job launcher that can be73        overridden by the user. Not applicable for JOB_LAUNCHER_NOBATCH and74        JOB_LAUNCHER_FAKE75    allow_missing_entries (bool): For a machine that generally requires certain entries76        (e.g., account): If allow_missing_entries is True, then we proceed even if these77        entries are missing. This is intended for when create_machine is just called for78        the sake of getting default values.79    """80    # ------------------------------------------------------------------------81    # Settings that are independent of both machine and job launcher type82    # ------------------------------------------------------------------------83    if account is None:84        account = _get_account()85    # ------------------------------------------------------------------------86    # Settings that depend on machine87    # ------------------------------------------------------------------------88    mach_defaults = defaults.get(machine_name)89    baseline_dir = None90    create_test_retry = 091    create_test_queue = CREATE_TEST_QUEUE_UNSPECIFIED92    if mach_defaults is not None:93        if job_launcher_type is None:94            job_launcher_type = mach_defaults.job_launcher_type95        if scratch_dir is None:96            scratch_dir = mach_defaults.scratch_dir97        # NOTE(wjs, 2019-05-17) Note that we don't provide a way to override the default98        # baseline_dir. The idea is that this should always provide the standard baseline99        # directory for this machine, even if a particular application wants to use100        # something different (in contrast to, say, scratch_dir, for which the value in101        # the machines object can itself be overridden). I think this will be smoother and102        # more intuitive if different baseline directories are needed for different103        # purposes in a single application (e.g., different baseline directories for104        # generation and comparison, or making a link in some temporary location that105        # points to the standard baselines).106        baseline_dir = mach_defaults.baseline_dir107        # We also don't provide a way to override the default create_test_retry or108        # create_test_queue in the machine object: these will always give the default109        # value for this machine, and other mechanisms will be given for overriding these110        # in a particular case.111        create_test_retry = mach_defaults.create_test_retry112        create_test_queue = mach_defaults.create_test_queue113        if account is None and mach_defaults.account_required and not allow_missing_entries:114            raise RuntimeError("Could not find an account code")115    else:116        if not allow_missing_entries:117            # This isn't exactly a missing entry, but the times we don't care about this118            # warning tend to be the same as the times when allow_missing_entries is true119            logger.warning(120                "machine %s not recognized; using generic no-batch settings",121                machine_name,122            )123        if job_launcher_type is None:124            job_launcher_type = JOB_LAUNCHER_NOBATCH125    # ------------------------------------------------------------------------126    # Settings that depend on both machine and job launcher type127    # ------------------------------------------------------------------------128    # Required args are ones that cannot be overridden by the user. So it doesn't make129    # sense to have these in the argument list for this function: they should only come130    # from the defaults structure. If the user wants to provide their own arguments, those131    # should be provided via extra_args.132    job_launcher_required_args = ""133    if mach_defaults is not None:134        these_defaults = mach_defaults.job_launcher_defaults.get(job_launcher_type)135        if these_defaults is not None:136            if job_launcher_queue is None:137                job_launcher_queue = these_defaults.queue138            if job_launcher_walltime is None:139                job_launcher_walltime = these_defaults.walltime140            if job_launcher_extra_args is None:141                job_launcher_extra_args = these_defaults.extra_args142            job_launcher_required_args = these_defaults.required_args143    # ------------------------------------------------------------------------144    # Create the job launcher and the full machine object145    # ------------------------------------------------------------------------146    job_launcher = create_job_launcher(147        job_launcher_type=job_launcher_type,148        account=account,149        queue=job_launcher_queue,150        walltime=job_launcher_walltime,151        nice_level=job_launcher_nice_level,152        required_args=job_launcher_required_args,153        extra_args=job_launcher_extra_args,154        allow_missing_entries=allow_missing_entries,155    )156    return Machine(157        name=machine_name,158        scratch_dir=scratch_dir,159        baseline_dir=baseline_dir,160        account=account,161        create_test_retry=create_test_retry,162        create_test_queue=create_test_queue,163        job_launcher=job_launcher,164    )165def get_possibly_overridden_mach_value(machine, varname, value=None):166    """Get the value to use for the given machine variable167    If value is provided (not None), use that. Otherwise use the value of the given168    variable from the provided machine object.169    Args:170    machine (Machine)171    varname (str): name of variable to get from the machine object172    value: if not None, use this instead of fetching from the machine object173    """174    if value is not None:175        return value176    return getattr(machine, varname)177def _get_account():178    account = get_project()...

machine_defaults.py

Source:machine_defaults.py

1"""Machine-specific default values.2To allow running out-of-the-box on other machines, add code here."""3from collections import namedtuple4import os5from ctsm.joblauncher.job_launcher_factory import JOB_LAUNCHER_QSUB6from ctsm.machine import CREATE_TEST_QUEUE_UNSPECIFIED7from ctsm.machine_utils import get_user8MachineDefaults = namedtuple(9    "MachineDefaults",10    [11        "job_launcher_type",12        "scratch_dir",13        "baseline_dir",14        "account_required",15        "create_test_retry",16        "create_test_queue",17        "job_launcher_defaults",18    ],19)20# job_launcher_type: one of the JOB_LAUNCHERs defined in job_launcher_factory21# scratch_dir: str22# baseline_dir: str: The standard location for CTSM baselines on this machine23# job_launcher_defaults: dict: keys are the JOB_LAUNCHERs defined in job_launcher_factory,24#     values are types defined here (like _QsubDefaults). A given machine's defaults can25#     have 0, 1 or multiple job_launcher_defaults. (It can be useful to have defaults even26#     for the non-default job launcher for this machine, in case the user chooses a27#     non-default launcher.)28# create_test_retry: int: Default number of times to retry a create_test job on this machine29# create_test_queue: str: Default queue to use for create_test; if this is30#     CREATE_TEST_QUEUE_UNSPECIFIED, then we won't add a '--queue' option to create_test,31#     instead leaving that value unspecified, allowing CIME to pick an appropriate queue32#     for each test using its standard mechanisms.33# account_required: bool: whether an account number is required on this machine (not34#     really a default, but used for error-checking)35# Note that the different job launcher types have different structures defining their36# defaults, because different ones require different elements to be set. For now we only37# have defaults for qsub, because other launchers (like no_batch) don't need any38# arguments.39QsubDefaults = namedtuple("QsubDefaults", ["queue", "walltime", "extra_args", "required_args"])40MACHINE_DEFAULTS = {41    "cheyenne": MachineDefaults(42        job_launcher_type=JOB_LAUNCHER_QSUB,43        scratch_dir=os.path.join(os.path.sep, "glade", "scratch", get_user()),44        baseline_dir=os.path.join(os.path.sep, "glade", "p", "cgd", "tss", "ctsm_baselines"),45        account_required=True,46        create_test_retry=0,47        # NOTE(wjs, 2022-02-23) By default, use the regular queue, even for48        # single-processor jobs. This is because the share queue has been really flaky,49        # with lots of job failures or slow-running jobs.50        create_test_queue="regular",51        job_launcher_defaults={52            JOB_LAUNCHER_QSUB: QsubDefaults(53                queue="regular",54                walltime="11:50:00",55                extra_args="",56                # The following assumes a single node, with a single mpi proc; we may want57                # to add more flexibility in the future, making the node / proc counts58                # individually selectable59                required_args="-l select=1:ncpus=36:mpiprocs=1 -V -r n -l inception=login -k oed",60            )61        },62    ),63    "hobart": MachineDefaults(64        job_launcher_type=JOB_LAUNCHER_QSUB,65        scratch_dir=os.path.join(os.path.sep, "scratch", "cluster", get_user()),66        baseline_dir=os.path.join(os.path.sep, "fs", "cgd", "csm", "ccsm_baselines"),67        account_required=False,68        create_test_retry=0,69        create_test_queue=CREATE_TEST_QUEUE_UNSPECIFIED,70        job_launcher_defaults={71            JOB_LAUNCHER_QSUB: QsubDefaults(72                queue="medium",73                walltime="04:00:00",74                extra_args="",75                required_args="-l nodes=1:ppn=48 -r n",76            )77        },78    ),79    "izumi": MachineDefaults(80        job_launcher_type=JOB_LAUNCHER_QSUB,81        scratch_dir=os.path.join(os.path.sep, "scratch", "cluster", get_user()),82        baseline_dir=os.path.join(os.path.sep, "fs", "cgd", "csm", "ccsm_baselines"),83        account_required=False,84        # jobs on izumi experience a high frequency of failures, often at the very end of85        # the job; so we'll automatically retry a failed job twice before giving up on it86        create_test_retry=2,87        create_test_queue=CREATE_TEST_QUEUE_UNSPECIFIED,88        job_launcher_defaults={89            JOB_LAUNCHER_QSUB: QsubDefaults(90                queue="medium",91                walltime="04:00:00",92                extra_args="",93                required_args="-l nodes=1:ppn=48 -r n",94            )95        },96    ),...

initialize_workload.py

Source:initialize_workload.py

1import sqlite32import configparser3import random4import json5import functions as x6import table_functions as tf7import shutil8from datetime import datetime910### Initialize Config ###11config = configparser.ConfigParser()12config.read('workload.ini')13create_test_queue= eval(config.get("query_queue", "create_test_queue"))14queue_settings= eval(config.get("query_queue", "queue_creation"))15backup_restore_query_files = eval(config.get("query_queue", "backup_restore_query_files"))16#print(query_settings)17query_folder= eval(config.get("general_settings", "query_folder"))18filter_choices=dict(config.items('query_filters'))1920# get execution mode ( production or development )21execution_mode= eval(config.get("general_settings", "mode"))22# get param values for the execution mode above23workload_db=eval(config.get(execution_mode+"_settings", "sqlite_db"))24query_frequency_file=eval(config.get(execution_mode+"_settings", "query_frequency_file"))2526dev=x.development_mode(execution_mode)2728### backup previous workload db29dateTimeObj = datetime.now()30timestampStr = dateTimeObj.strftime("%Y%m%d_%H%M%S")31shutil.copy(workload_db, workload_db+'.'+timestampStr)3233### initialize sqlite34[conn, cursor]=x.sqlite3db(workload_db)3536# reads query names from csv, read query SQL from query folder, searches required filters in query sql37queries=x.load_query_files(query_frequency_file)38x.populate_workload_frequency_table(workload_db,queries)3940tf.create_table_workload_queues(workload_db)41tf.create_table_workload_log(workload_db)4243#create each queue44for queueid, values in queue_settings.items():45 duration=values['duration']46 frequency_multiplier=values['freq']47 x.dev_print(dev,'QUEUEID: '+queueid)48 #print(duration)49 50 #create execution times for each query51 for index, row in queries.iterrows():52   x.dev_print(dev,row['file']+str(row['frequency']))53   query_file=row['file']54   with open(query_folder+'/'+query_file+'.sql', 'r') as myfile:55    original_query_text = myfile.read()56   hour=057   #for each duration value (each hour)58   while hour < duration :59    count=060    while count < row['frequency']*frequency_multiplier:61     count+=162     start_time=random.randrange(0,60,1)+(hour*60)63     x.dev_print(dev,'time: '+str(start_time))64     fields={}65     for filter in row['filters']:66       #print(filter)67       value_list=filter_choices[filter].split(',')68       if "date" not in filter:69        # between 1 and 4 values70        number_of_values=random.randrange(1,5)71        filter_values=random.sample(value_list,k=number_of_values)72        filter_value = ','.join(map(str, filter_values))73        #print(type(filter_values))74        #print(random.sample(value_list,k=2))75       else:76        filter_value=random.choice(value_list)77       #print(filter_value)78       fields[filter]=filter_value79     x.dev_print(dev,fields)80     #print(original_query_text.format(**fields))81     filtered_query_text=original_query_text.format(**fields)     82     #print(type(fields))83     json_fields=json.dumps(fields)84     cursor.execute("insert into workload_queues (queue, execution_minute, query_file,filter_values) values ( ?,?,?,?)",(queueid,start_time,row['file'],json_fields))85    hour+=186conn.commit() 8788if create_test_queue == 't' :89  cursor.execute("insert into workload_queues (queue, execution_minute, query_file,filter_values) select 'q0', 0, query_file, min(filter_values) from workload_queues group by 1,2,3;")9091#create backup/restore queue92start_minute=093no_filters={}94for query_file in backup_restore_query_files:95  cursor.execute("insert into workload_queues (queue, execution_minute, query_file,filter_values) values (?,?,?,?)", ('q_br', start_minute, query_file,json.dumps(no_filters)))96  cursor.execute("insert into workload_frequencies ( query, parallel_run, only_sequential) values (?,?,?)",(query_file,0,1))97  start_minute+=19899conn.commit()100cursor.close()
...

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.