TestMu Conference is Back! Future of Testing + AIRegister for Free

How to use task_priority_key method in autotest

Best Python code snippet using autotest_python

query_managers.py

Source:query_managers.py

...127                               models.SpecialTask.Task.CLEANUP,128                               models.SpecialTask.Task.VERIFY,129                               models.SpecialTask.Task.RESET,130                               models.SpecialTask.Task.PROVISION]131        def task_priority_key(task):132            return task_priority_order.index(task.task)133        return sorted(queued_tasks, key=task_priority_key)134    @classmethod135    def get_overlapping_jobs(cls):136        """A helper method to get all active jobs using the same host.137        @return: A list of dictionaries with the hqe id, job_id and host_id138            of the currently overlapping jobs.139        """140        # Filter all active hqes and stand alone special tasks to make sure141        # a host isn't being used by two jobs at the same time. An incomplete142        # stand alone special task can share a host with an active hqe, an143        # example of this is the cleanup scheduled in gathering.144        hqe_hosts = list(models.HostQueueEntry.objects.filter(145                active=1, complete=0, host_id__isnull=False).values_list(...

sls.py

Source:sls.py

1#!/usr/bin/env python2import re3from json import JSONDecoder4from operator import attrgetter5from models.yarn.penalties import get_penalty, YarnPenalty6from models.yarn.objects import *7from utils import lines_per_n8LINES_TO_READ = 109TASK_NR_KEY = "c.nr"10TASK_DURATION_KEY = "c.dur"11TASK_MEMORY_KEY = "c.mem"12TASK_CORES_KEY = "c.cores"13TASK_PRIORITY_KEY = "c.prio"14TASK_TYPE_KEY = "c.type"15TASK_PENALTY_KEY = "c.penalty"16TASK_IB_KEY = "c.ib"17JOB_AM_TYPE_KEY = "am.type"18JOB_ID_KEY = "job.id"19JOB_TASKS_KEY = "job.tasks"20JOB_START_MS_KEY = "job.start.ms"21JOB_END_MS_KEY = "job.end.ms"22RACK_NAME_KEY = "rack"23RACK_NODES_KEY = "nodes"24NODE_KEY = "node"25LOG = logging.getLogger('sls_parser')26class SLSParser(object):27    def __init__(self, sls_file, topo_file, node_mem_capacity, node_core_capacity, node_hb_ms, am_hb_ms,28                 am_container_mb, am_container_cores, use_meganode, default_task_penalty):29        self.sls_file = sls_file30        self.topo_file = topo_file31        self.am_container_resource = YarnResource(am_container_mb, am_container_cores)32        self.node_resource = YarnResource(memory_mb=node_mem_capacity, vcores=node_core_capacity)33        self.am_hb_ms = am_hb_ms34        self.node_hb_ms = node_hb_ms35        self.use_meganode = use_meganode36        self.default_task_penalty = default_task_penalty37    @staticmethod38    def _print_chunk(chunk):39        for j, line in enumerate(chunk.splitlines()):40            print '{0:<5}{1}'.format(j+1, line)41    def parse_topo(self):42        """ Parse a YARN SLS topology file. This is a JSON file containing multiple rack configurations. """43        json_decoder = JSONDecoder()44        rack_objects = []45        with open(self.topo_file) as topo_file:46            lines = "".join(topo_file.readlines()).strip()47            done_parsing_file = False48            while not done_parsing_file:49                try:50                    rack_object, object_end = json_decoder.raw_decode(lines)51                except ValueError as e:52                    LOG.exception("Unable to parse topology file", exc_info=e)53                    break54                rack_objects.append(rack_object)55                if object_end != len(lines):56                    lines = lines[object_end + 1:]57                else:58                    done_parsing_file = True59        return rack_objects60    def get_yarn_topo(self):61        """ Parse a YARN SLS topology file and return a touple (racks, nodes).62        'racks' are a dict of YarnRack objects, 'nodes' are a dict of YarnNode objects. """63        racks = {}64        nodes = {}65        rack_objects = self.parse_topo()66        node_id_counter = 167        for rack_object in rack_objects:68            rack = YarnRack(rack_object[RACK_NAME_KEY])69            if self.use_meganode:70                # Generate racks with 1 node that has all the71                # resources pooled72                resource = YarnResource(self.node_resource.memory_mb * len(rack_object[RACK_NODES_KEY]),73                                        self.node_resource.vcores * len(rack_object[RACK_NODES_KEY]))74                name = "meganode1"75                node = YarnNode(name, resource, rack, self.node_hb_ms, node_id=node_id_counter)76                node_id_counter += 177                rack.add_node(node)78                nodes[name] = node79            else:80                for node_object in rack_object[RACK_NODES_KEY]:81                    node = YarnNode(82                            name=node_object[NODE_KEY],83                            resource=copy.copy(self.node_resource),84                            rack=rack,85                            hb_interval_ms=self.node_hb_ms,86                            node_id=node_id_counter)87                    node_id_counter += 188                    rack.add_node(node)89                    nodes[node.name] = node90            racks[rack.name] = rack91        return racks, nodes92    def parse_sls(self):93        """ Parse a YARN SLS trace file. This is a JSON file containing multiple job objects. """94        json_decoder = JSONDecoder()95        job_objects = []96        value_error_pattern = re.compile('Expecting .+ \(char (\d+)\)$')97        with open(self.sls_file) as sls_file:98            object_chunk = ''99            last_error_idx = -1100            # Read file in chunks of lines.101            for chunk in lines_per_n(sls_file, LINES_TO_READ):102                # Remove all whitespace103                chunk = chunk.replace(" ", "")104                chunk = chunk.replace("\n", "")105                # Add (hopefully good) whitespace106                chunk = re.sub(r"{", r'{\n', chunk)107                chunk = re.sub(r"}", r'}\n', chunk)108                chunk = re.sub(r"\[", r'[\n', chunk)109                chunk = re.sub(r"\]", r']\n', chunk)110                # Further sanitize some JSON stuff111                chunk = re.sub(r"{\s*'?(\w)", r'{"\1', chunk)112                chunk = re.sub(r",\s*'?(\w)", r',"\1', chunk)113                chunk = re.sub(r"(\w)'?\s*:", r'\1":', chunk)114                chunk = re.sub(r":\s*'(\w+)'\s*([,}])", r':"\1"\2', chunk)115                object_chunk += chunk116                # Try to parse chunk read so far.117                chunk_parsing_done = False118                # Chunk may contain more than one object.119                while not chunk_parsing_done:120                    try:121                        parse_result = json_decoder.raw_decode(object_chunk)122                        last_error_idx = -1123                    except ValueError as e:124                        m = value_error_pattern.match(e.message)125                        if m:126                            # Get the index that the parsing error occurred on.127                            idx = int(m.group(1))128                            if last_error_idx == -1 or last_error_idx != idx:129                                # Chunk is not yet complete, keep reading.130                                last_error_idx = idx131                                break132                        # The error at the current index was not due to an incomplete chunk.133                        SLSParser._print_chunk(object_chunk)134                        raise e135                    # Add decoded job object to array136                    job_objects.append(parse_result[0])137                    # Check if there's trailing data from another object138                    object_end = parse_result[1]139                    if object_end != len(object_chunk):140                        # Trim chunk for the next object141                        object_chunk = object_chunk[object_end + 1:]142                    if not object_chunk.isspace():143                        chunk_parsing_done = True144        return job_objects145    def get_yarn_jobs(self):146        """ Parse the SLS trace file and return a list of YarnJob objects. """147        jobs = self.parse_sls()148        yarn_jobs = []149        for job in jobs:150            yarn_tasks = []151            # Generate the job object152            job_name = job[JOB_ID_KEY]153            # Translate "job_XXX" to an int154            job_id = int(job_name[4:])155            start_ms = -1156            start_after_job = None157            if type(job[JOB_START_MS_KEY]) is int:158                start_ms = job[JOB_START_MS_KEY]159            else:160                start_after_job = int(job[JOB_START_MS_KEY][4:])161            yarn_j = YarnJob(162                    am_type=YarnAMType.__members__.get(job[JOB_AM_TYPE_KEY].upper()),163                    name=job_name,164                    job_id=job_id,165                    start_ms=start_ms,166                    end_ms=job[JOB_END_MS_KEY],167                    am_hb_ms=self.am_hb_ms,168                    tasks=yarn_tasks,169                    after_job=start_after_job)170            # Generate an AM container171            am_container = YarnPrototypeContainer(172                    num_containers=1,173                    resource=copy.copy(self.am_container_resource),174                    priority=0,175                    container_type=YarnContainerType.MRAM,176                    job=yarn_j)177            yarn_tasks.append(am_container)178            for task in job[JOB_TASKS_KEY]:179                # Generate all the other containers180                task_penalty = self.default_task_penalty181                if TASK_PENALTY_KEY in task:182                    if TASK_IB_KEY not in task:183                        LOG.warning("Task " + str(task) + " has a penalty model defined, but no IB set. Ignoring.")184                    else:185                        task_penalty = get_penalty(YarnPenalty.__members__.get(task[TASK_PENALTY_KEY].upper()),186                                                   initial_bump=float(task[TASK_IB_KEY]))187                yarn_task = YarnPrototypeContainer(188                        num_containers=int(task[TASK_NR_KEY]),189                        duration=int(task[TASK_DURATION_KEY]),190                        resource=YarnResource(memory_mb=int(task[TASK_MEMORY_KEY]),191                                              vcores=int(task.get(TASK_CORES_KEY, '1'))),192                        priority=int(task[TASK_PRIORITY_KEY]),193                        container_type=YarnContainerType.__members__.get(task[TASK_TYPE_KEY].upper()),194                        job=yarn_j,195                        penalty=task_penalty)196                yarn_tasks.append(yarn_task)197            # Sort containers by priority198            yarn_tasks.sort(key=attrgetter('priority'))199            yarn_jobs.append(yarn_j)...

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.