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

How to use __resolve method in yandex-tank

Best Python code snippet using yandex-tank

flow.py

Source:flow.py

...16        self.byte_count = 017        self.throughput = 018        self.timestamp = time.time()19        self.path = path20    def __resolve(self, match, key):21        try:22            ret = match[key]23        except KeyError:24            ret = None25        return ret26    def __eq__(self, other):27        is_equal = True28        # check match.29        we = self.__resolve(self.match, 'eth_dst')30        they = self.__resolve(other.match, 'eth_dst')31        is_equal = is_equal and (we == they)32        we = self.__resolve(self.match, 'eth_src')33        they = self.__resolve(other.match, 'eth_src')34        is_equal = is_equal and (we == they)35        we = self.__resolve(self.match, 'in_port')36        they = self.__resolve(other.match, 'in_port')37        is_equal = is_equal and (we == they)38        we = self.__resolve(self.match, 'eth_type')39        they = self.__resolve(other.match, 'eth_type')40        is_equal = is_equal and (we == they)41        we = self.__resolve(self.match, 'ipv4_src')42        they = self.__resolve(other.match, 'ipv4_src')43        is_equal = is_equal and (we == they)44        we = self.__resolve(self.match, 'ipv4_dst')45        they = self.__resolve(other.match, 'ipv4_dst')46        is_equal = is_equal and (we == they)47        we = self.__resolve(self.match, 'ip_proto')48        they = self.__resolve(other.match, 'ip_proto')49        is_equal = is_equal and (we == they)50        we = self.__resolve(self.match, 'udp_src')51        they = self.__resolve(other.match, 'udp_src')52        is_equal = is_equal and (we == they)53        we = self.__resolve(self.match, 'udp_dst')54        they = self.__resolve(other.match, 'udp_dst')55        is_equal = is_equal and (we == they)56        we = self.__resolve(self.match, 'tcp_src')57        they = self.__resolve(other.match, 'tcp_src')58        is_equal = is_equal and (we == they)59        we = self.__resolve(self.match, 'tcp_dst')60        they = self.__resolve(other.match, 'tcp_dst')61        is_equal = is_equal and (we == they)62        # check actions.63        try:64            we = self.actions[0].port if self.actions else None65        except AttributeError:66            we = None67        try:68            they = other.actions[0].port if other.actions else None69        except AttributeError:70            they = None71        is_equal = is_equal and (we == they)72        is_equal = is_equal and (self.idle_timeout == other.idle_timeout)73        is_equal = is_equal and (self.hard_timeout == other.hard_timeout)74        is_equal = is_equal and (self.priority == other.priority)75        return is_equal == True76    def __str__(self):77        header =  'packets  '78        header += 'bytes    '79        header += 'eth-dst           '80        header += 'eth-src           '81        header += 'in-port  '82        header += 'out-port '83        header += 'throughput   '84        header += '\n'85        header += ('-' * 8) + ' ' + ('-' * 8) + ' '86        header += ('-' * 17) + ' ' + ('-' * 17) + ' '87        header += ('-' * 8) + ' ' + ('-' * 8) + ' '88        header += ('-' * 12)89        info =  '%8d' % self.packet_count90        info += '|%8d' % self.byte_count91        dst      = self.__resolve(self.match, 'eth_dst')92        src      = self.__resolve(self.match, 'eth_src')93        in_port  = self.__resolve(self.match, 'in_port')94        try:95            out_port = self.actions[0].port if self.actions else None96        except AttributeError:97            out_port = None98        info += '|%017s' % dst if dst is not None else '|%017s' % 'unspec.'99        info += '|%017s' % src if src is not None else '|%017s' % 'unspec.'100        info += '|%8d' % in_port if in_port is not None else '|%08s' % 'unspec.'101        info += '|%8d' % out_port if out_port is not None else '|%08s' % 'unspec.'102        info += '|%12d' % self.throughput103        info += '\n'104        return header + '\n' + info105    def update_stats(self, packet_count, byte_count):106        now = time.time()107        interval = now - self.timestamp...

Resolver.py

Source:Resolver.py

...10        self.scopes = []11        self.__currentFunction = FunctionType.NONE12    def resolve(self, stmts: list[Stmt]) -> None:13        for stmt in stmts:14            self.__resolve(stmt)15    16    def __resolve(self, stmt: Stmt|Expr) -> None:17        stmt.accept(self)18    # override19    def visitBlock(self, stmt: Block) -> None:20        self.__beginScope()21        self.resolve(stmt.statements)22        self.__endScope()23        return24    25    def __beginScope(self) -> None:26        self.scopes.append({})27    28    def __endScope(self) -> None:29        self.scopes.pop()30    31    #override32    def visitVarVarDecl(self, stmt: VarDecl) -> None:33        self.__declare(stmt.name)34        if (stmt.indentifier != None):35            self.__resolve(stmt.indentifier)36        self.__define(stmt.name)37        return38    def __declare(self, name: Token) -> None:39        if (len(self.scopes) == 0):40            return41        scope = self.scopes[-1]42        if (name.lexeme in scope):43            ErrorHandler().error(name, "Variable with this name already in this scope.")44        scope[name.lexeme] = False45    46    def __define(self, name: Token) -> None:47        if (len(self.scopes) == 0):48            return49        scope = self.scopes[-1]50        scope[name.lexeme] = True51    52    #override53    def visitVariableExpr(self, expr: Variable) -> None:54        if (len(self.scopes) != 0) and (self.scopes[-1].get(expr.name.lexeme) == False):55            raise "Can't read local variable in its own initializer."56        self.__resolveLocal(expr, expr.name)57        return58    59    def __resolveLocal(self, expr: Expr, name: Token) -> None:60        for idx in range(len(self.scopes)-1, -1, -1):61            if (name.lexeme in self.scopes[idx]):62                self.interpreter.resolve(expr, len(self.scopes)-1-idx)63                return64        65    #override66    def visitAssignExpr(self, expr: Assign) -> None:67        self.__resolve(expr.val)68        self.__resolveLocal(expr, expr.name)69        return70    71    #override72    def visitFuncStmt(self, stmt: FuncStmt) -> None:73        self.__declare(stmt.name)74        self.__define(stmt.name)75        self.__resolveFunc(stmt, FunctionType.FUNCTION)76    77    def __resolveFunc(self, func: FuncStmt, type: FunctionType) -> None:78        enclosingFunction = self.__currentFunction79        self.__currentFunction = type80        self.__beginScope()81        for param in func.params:82            self.__declare(param)83            self.__define(param)84        self.resolve(func.body)85        self.__endScope()86        self.__currentFunction = enclosingFunction87    88    #override89    def visitExprStm(self, stmt: ExprStmt) -> None:90        self.__resolve(stmt.expr)91        return92    93    #override94    def visitIF(self, stmt: IF) -> None:95        self.__resolve(stmt.condition)96        self.__resolve(stmt.thenBranch)97        if (stmt.elseBranch):98            self.__resolve(stmt.elseBranch)99        return100    101    #override102    def visitPrintStmt(self, stmt: PrintStmt) -> None:103        self.__resolve(stmt.expr)104        return105    106    #override107    def visitReturnStmt(self, stmt: ReturnStmt) -> None:108        if (self.__currentFunction == FunctionType.NONE):109            ErrorHandler().error(stmt.keyword, "Variable with this name already in this scope.")110        if (stmt.val != None):111            self.__resolve(stmt.val)112        return113    #override114    def visitWhileStmt(self, stmt: WhileStmt) -> None:115        self.__resolve(stmt.condition)116        self.__resolve(stmt.body)117        return118    119    #override120    def visitBinaryExpr(self, expr: Binary) -> None:121        self.__resolve(expr.left)122        self.__resolve(expr.right)123        return124    125    #override126    def visitCallExpr(self, expr: Call) -> None:127        self.__resolve(expr.callee)128        for arg in expr.arguments:129            self.__resolve(arg)130        return131    132    #override133    def visitGroupExpr(self, expr: Group) -> None:134        self.__resolve(expr.expression)135        return136    137    #override138    def visitLiteralExpr(self, expr: Literal) -> None:139        return140    141    #override142    def visitLogicalExpr(self, expr: Logical):143        self.__resolve(expr.left)144        self.__resolve(expr.right)145        return146    147    #override148    def visitUnaryExpr(self, expr: Unary):149        self.__resolve(expr.right)150        return151    ...

environment.py

Source:environment.py

...7    def define(self, name, value):8        self.record[name] = value9        return value10    def assign(self, name, value):11        self.__resolve(name).record[name] = value12        return value13    def lookup(self, name):14        return self.__resolve(name).record[name]15    def __resolve(self, name):16        if name in self.record:17            return self18        elif self.parent is None:19            raise KeyError('Cannot find variable name: ', name);20        else:...

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.