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

How to use lock_duration method in yandex-tank

Best Python code snippet using yandex-tank

cell_graph_locker.py

Source:cell_graph_locker.py

1import numpy as np2class CellGraphLocker:3    def __init__(self, graph):4        self.graph = graph5        self.data = []   # list of list of listï¼é¿åº¦ä¸ºvertexæ°éï¼æ¯ä¸ä¸ªvertexä¸åå«æç§ååé¡ºåºæåå¥½ç(duration, agent_idx)6        self.reset()7    def reset(self):8        vertexes = len(self.graph.vertexes)9        self.data = [[] for i in range(vertexes)]   # list of list10    def lock(self, vertex_idx, agent_idx, duration):       # æ¯ä¸æ¬¡locké½æ´æ°self.data11        # assert not self.is_locked(vertex_idx, agent_idx, duration)12        if len(self.data[vertex_idx])==0:13            self.data[vertex_idx].append((duration, agent_idx))14            return15        # index = self.equal_or_greater_index(vertex_idx, duration[0])16        index = self.equal_or_greater_index_end(vertex_idx, duration[1])  # duration[0]æ¯å°è¾¾æ¶é´ï¼1æ¯ä»è¿ä¸ªç¹çåºåæ¶é´17        if index < len(self.data[vertex_idx]):18            curr_lock_info = self.data[vertex_idx][index]19            if (curr_lock_info[1] == agent_idx) and self._has_intersection(curr_lock_info[0], duration):20                assert (curr_lock_info[0][0] <= duration[0]) and (duration[1] <= curr_lock_info[0][1])21                return22            assert curr_lock_info[0][0] >= duration[1]23            self.data[vertex_idx].insert(index, (duration, agent_idx))24            # if (curr_lock_info[1]==agent_idx) and (curr_lock_info[0][1] == duration[1]) and (curr_lock_info[0][0] <= duration[0]):25            #     self.data[vertex_idx][index] = (duration, agent_idx)26            # else:27            #     self.data[vertex_idx].insert(index, (duration, agent_idx) )28        else:29            self.data[vertex_idx].append((duration, agent_idx))30    def is_locked(self, vertex_idx, agent_idx, duration):31        if len(self.data[vertex_idx])==0:32            return False33        new_lock = (duration, agent_idx)34        left_lock = None35        right_lock = None36        index = self.equal_or_greater_index(vertex_idx, duration[0])37        if index < len(self.data[vertex_idx]):38            if self.data[vertex_idx][index][0][0] == duration:39                return True40            right_lock = self.data[vertex_idx][index]41            if index>0:42                left_lock = self.data[vertex_idx][index - 1]43        else:44            left_lock = self.data[vertex_idx][index - 1]45        return self._has_conflict(left_lock, new_lock) or self._has_conflict(new_lock, right_lock)46        # index = self.equal_or_greater_index(vertex_idx, duration[0])47        # if index < len(self.data[vertex_idx]):48        #     lock_duration, lock_agent_idx = self.data[vertex_idx][index]49        #     if (lock_duration[0] < duration[1]) and (agent_idx != lock_agent_idx):50        #         return True51        #52        # if index > 0:53        #     lock_duration, lock_agent_idx = self.data[vertex_idx][index - 1]54        #     if (lock_duration[1] > duration[0]) and (agent_idx != lock_agent_idx):55        #         return True56        #57        # return False58    def _has_conflict(self, left, right):          # leftårighté½æ¯[duration,agent_index]59        if (left is None) or (right is None):60            return False61        d1 = left[0]   # å·¦è¾¹duration62        d2 = right[0]   # å³è¾¹duration63        if left[1] > right[1]:   64            d1 = (d1[0], d1[1] + 1)65        return self._has_intersection(d1, d2)   # d1åd2è¦æ¯æäº¤éåä¼å²çª66    def next_free_time(self, vertex_idx, agent_idx, duration):67        index = self.equal_or_greater_index_end(vertex_idx, duration[1])68        if index < len(self.data[vertex_idx]):69            # lock_duration = self.data[vertex_idx][index][0]70            # if self._has_intersection(duration, lock_duration):71            #     return lock_duration[1]72            if self._has_conflict((duration, agent_idx), self.data[vertex_idx][index]):73                return self.data[vertex_idx][index][0][1]74        if index > 0:75            # lock_duration = self.data[vertex_idx][index - 1][0]76            # if self._has_intersection(duration, lock_duration):77            #     return lock_duration[1]78            if self._has_conflict(self.data[vertex_idx][index-1], (duration, agent_idx)):79                return self.data[vertex_idx][index-1][0][1]80        # print('already free')81        return duration[0]82        # index = self.equal_or_greater_index(vertex_idx, duration[0])83        # if index < len(self.data[vertex_idx]):84        #     lock_duration, lock_agent_idx = self.data[vertex_idx][index]85        #     if (lock_duration[0] < duration[1]) and (agent_idx != lock_agent_idx):86        #         return True87        #88        # if index > 0:89        #     lock_duration, lock_agent_idx = self.data[vertex_idx][index - 1]90        #     if (lock_duration[1] > duration[0]) and (agent_idx != lock_agent_idx):91        #         return True92        #93        # return False94    def unlock(self, vertex_idx, agent_idx, duration):95        assert len(self.data[vertex_idx])96        index = self.equal_or_greater_index(vertex_idx, duration[0])97        assert (index >= 0) and (index < len(self.data[vertex_idx]))98        lock_duration, lock_agent_idx = self.data[vertex_idx][index]99        assert (lock_duration == duration) and (lock_agent_idx == agent_idx)100        self.data[vertex_idx].pop(index)101    def equal_or_greater_index(self, vertex_idx, start_time):102        # d = self.data[vertex_idx]103        #104        # if not len(d):105        #     return 0106        #107        # l = 0108        # r = len(d) - 1109        #110        # while l <= r:111        #     c = (l + r) // 2112        #113        #     lock_duration_start = d[c][0][0]114        #     if lock_duration_start == start_time:115        #         return c116        #     elif lock_duration_start < start_time:117        #         l = c + 1118        #     else:119        #         r = c - 1120        #121        # return max(l, r)122        #123        for i, (lock_duration, lock_agent_idx) in enumerate(self.data[vertex_idx]):124            if lock_duration[0] >= start_time:125                return i126        return len(self.data[vertex_idx])127    def equal_or_greater_index_end(self, vertex_idx, end_time):128        for i, (lock_duration, lock_agent_idx) in enumerate(self.data[vertex_idx]):129            if lock_duration[1] >= end_time:130                return i131        return len(self.data[vertex_idx])132    def _has_intersection(self, a, b):133        return not ((a[1] <= b[0]) or (b[1] <= a[0]))134    def unlock_agent(self, agent_id):135        for i in range(len(self.data)):136            for j in reversed(range(len(self.data[i]))):137                if self.data[i][j][1] == agent_id:138                    self.data[i].pop(j)139    def unlock_agent_with_list(self, agent_id, vertex_list):140        for i in vertex_list:141            for j in reversed(range(len(self.data[i]))):142                if self.data[i][j][1] == agent_id:143                    self.data[i].pop(j)144    def last_time_step(self, vertex_idx, agent_idx):145        if not len(self.data[vertex_idx]):146            return 0147        res = self.data[vertex_idx][-1][0][1]148        if self.data[vertex_idx][-1][1] != agent_idx:149            res += 1...

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.