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

How to use read_token method in yandex-tank

Best Python code snippet using yandex-tank

TypeScriptParser.py

Source:TypeScriptParser.py

...50            type_args = self.parse_type_args()51            self.reader.expect_token("(")52            args = self.expression_parser.parse_call_arguments()53            return exprs.UnresolvedCallExpression(left, type_args, args)54        elif self.reader.read_token("instanceof"):55            type = self.parse_type()56            return exprs.InstanceOfExpression(left, type)57        elif isinstance(left, exprs.Identifier) and self.reader.read_token("=>"):58            block = self.parse_lambda_block()59            return types.Lambda([types.MethodParameter(left.text, None, None, None)], block)60        return None61    62    def parse_lambda_params(self):63        if not self.reader.read_token("("):64            return None65        66        params = []67        if not self.reader.read_token(")"):68            while True:69                param_name = self.reader.expect_identifier()70                type = self.parse_type() if self.reader.read_token(":") else None71                params.append(types.MethodParameter(param_name, type, None, None))72                if not (self.reader.read_token(",")):73                    break74            self.reader.expect_token(")")75        return params76    77    def parse_type(self):78        if self.reader.read_token("{"):79            self.reader.expect_token("[")80            self.reader.read_identifier()81            self.reader.expect_token(":")82            self.reader.expect_token("string")83            self.reader.expect_token("]")84            self.reader.expect_token(":")85            map_value_type = self.parse_type()86            self.reader.read_token(";")87            self.reader.expect_token("}")88            return astTypes.UnresolvedType("TsMap", [map_value_type])89        90        if self.reader.peek_token("("):91            params = self.parse_lambda_params()92            self.reader.expect_token("=>")93            return_type = self.parse_type()94            return astTypes.LambdaType(params, return_type)95        96        type_name = self.reader.expect_identifier()97        start_pos = self.reader.prev_token_offset98        99        100        if type_name == "string":101            type = astTypes.UnresolvedType("TsString", [])102        elif type_name == "boolean":103            type = astTypes.UnresolvedType("TsBoolean", [])104        elif type_name == "number":105            type = astTypes.UnresolvedType("TsNumber", [])106        elif type_name == "any":107            type = astTypes.AnyType.instance108        elif type_name == "void":109            type = astTypes.VoidType.instance110        else:111            type_arguments = self.parse_type_args()112            type = astTypes.UnresolvedType(type_name, type_arguments)113        114        self.node_manager.add_node(type, start_pos)115        116        while self.reader.read_token("[]"):117            type = astTypes.UnresolvedType("TsArray", [type])118            self.node_manager.add_node(type, start_pos)119        120        return type121    122    def parse_expression(self):123        return self.expression_parser.parse()124    125    def unary_prehook(self):126        if self.reader.read_token("null"):127            return exprs.NullLiteral()128        elif self.reader.read_token("true"):129            return exprs.BooleanLiteral(True)130        elif self.reader.read_token("false"):131            return exprs.BooleanLiteral(False)132        elif self.reader.read_token("`"):133            parts = []134            lit_part = ""135            while True:136                if self.reader.read_exactly("`"):137                    if lit_part != "":138                        parts.append(exprs.TemplateStringPart.literal(lit_part))139                        lit_part = ""140                    141                    break142                elif self.reader.read_exactly("${"):143                    if lit_part != "":144                        parts.append(exprs.TemplateStringPart.literal(lit_part))145                        lit_part = ""146                    147                    expr = self.parse_expression()148                    parts.append(exprs.TemplateStringPart.expression(expr))149                    self.reader.expect_token("}")150                elif self.allow_dollar_ids and self.reader.read_exactly("$"):151                    if lit_part != "":152                        parts.append(exprs.TemplateStringPart.literal(lit_part))153                        lit_part = ""154                    155                    id = self.reader.read_identifier()156                    parts.append(exprs.TemplateStringPart.expression(exprs.Identifier(id)))157                elif self.reader.read_exactly("\\"):158                    chr = self.reader.read_char()159                    if chr == "n":160                        lit_part += "\n"161                    elif chr == "r":162                        lit_part += "\r"163                    elif chr == "t":164                        lit_part += "\t"165                    elif chr == "`":166                        lit_part += "`"167                    elif chr == "$":168                        lit_part += "$"169                    elif chr == "\\":170                        lit_part += "\\"171                    else:172                        self.reader.fail("invalid escape", self.reader.offset - 1)173                else:174                    chr = self.reader.read_char()175                    chr_code = ord(chr[0])176                    if not (32 <= chr_code and chr_code <= 126) or chr == "`" or chr == "\\":177                        self.reader.fail(f'''not allowed character (code={chr_code})''', self.reader.offset - 1)178                    lit_part += chr179            return exprs.TemplateString(parts)180        elif self.reader.read_token("new"):181            type = self.parse_type()182            if isinstance(type, astTypes.UnresolvedType):183                self.reader.expect_token("(")184                args = self.expression_parser.parse_call_arguments()185                return exprs.UnresolvedNewExpression(type, args)186            else:187                raise Error(f'''[TypeScriptParser2] Expected UnresolvedType here!''')188        elif self.reader.read_token("<"):189            new_type = self.parse_type()190            self.reader.expect_token(">")191            expression = self.parse_expression()192            return exprs.CastExpression(new_type, expression)193        elif self.reader.read_token("/"):194            pattern = ""195            while True:196                chr = self.reader.read_char()197                if chr == "\\":198                    chr2 = self.reader.read_char()199                    pattern += "/" if chr2 == "/" else "\\" + chr2200                elif chr == "/":201                    break202                else:203                    pattern += chr204            modifiers = self.reader.read_modifiers(["g", "i"])205            return exprs.RegexLiteral(pattern, "i" in modifiers, "g" in modifiers)206        elif self.reader.read_token("typeof"):207            expr = self.expression_parser.parse(self.expression_parser.prefix_precedence)208            self.reader.expect_token("===")209            check = self.reader.expect_string()210            211            ts_type = None212            if check == "string":213                ts_type = "TsString"214            elif check == "boolean":215                ts_type = "TsBoolean"216            elif check == "object":217                ts_type = "Object"218            elif check == "function":219                # TODO: ???220                ts_type = "Function"221            elif check == "undefined":222                # TODO: ???223                ts_type = "Object"224            else:225                self.reader.fail("unexpected typeof comparison")226            227            return exprs.InstanceOfExpression(expr, astTypes.UnresolvedType(ts_type, []))228        elif self.reader.peek_regex("\\([A-Za-z0-9_]+\\s*[:,]|\\(\\)") != None:229            params = self.parse_lambda_params()230            self.reader.expect_token("=>")231            block = self.parse_lambda_block()232            return types.Lambda(params, block)233        elif self.reader.read_token("await"):234            expression = self.parse_expression()235            return exprs.AwaitExpression(expression)236        237        map_literal = self.expression_parser.parse_map_literal()238        if map_literal != None:239            return map_literal240        241        array_literal = self.expression_parser.parse_array_literal()242        if array_literal != None:243            return array_literal244        245        return None246    247    def parse_lambda_block(self):248        block = self.parse_block()249        if block != None:250            return block251        252        return_expr = self.parse_expression()253        if isinstance(return_expr, exprs.ParenthesizedExpression):254            return_expr = return_expr.expression255        return stats.Block([stats.ReturnStatement(return_expr)])256    257    def parse_type_and_init(self):258        type = self.parse_type() if self.reader.read_token(":") else None259        init = self.parse_expression() if self.reader.read_token("=") else None260        261        if type == None and init == None:262            self.reader.fail(f'''expected type declaration or initializer''')263        264        return TypeAndInit(type, init)265    266    def expect_block_or_statement(self):267        block = self.parse_block()268        if block != None:269            return block270        271        stmts = []272        stmt = self.expect_statement()273        if stmt != None:274            stmts.append(stmt)275        return stats.Block(stmts)276    277    def expect_statement(self):278        statement = None279        280        leading_trivia = self.reader.read_leading_trivia()281        start_pos = self.reader.offset282        283        requires_closing = True284        var_decl_matches = self.reader.read_regex("(const|let|var)\\b")285        if var_decl_matches != None:286            name = self.reader.expect_identifier("expected variable name")287            type_and_init = self.parse_type_and_init()288            statement = stats.VariableDeclaration(name, type_and_init.type, type_and_init.init)289        elif self.reader.read_token("delete"):290            statement = stats.UnsetStatement(self.parse_expression())291        elif self.reader.read_token("if"):292            requires_closing = False293            self.reader.expect_token("(")294            condition = self.parse_expression()295            self.reader.expect_token(")")296            then = self.expect_block_or_statement()297            else_ = self.expect_block_or_statement() if self.reader.read_token("else") else None298            statement = stats.IfStatement(condition, then, else_)299        elif self.reader.read_token("while"):300            requires_closing = False301            self.reader.expect_token("(")302            condition = self.parse_expression()303            self.reader.expect_token(")")304            body = self.expect_block_or_statement()305            statement = stats.WhileStatement(condition, body)306        elif self.reader.read_token("do"):307            requires_closing = False308            body = self.expect_block_or_statement()309            self.reader.expect_token("while")310            self.reader.expect_token("(")311            condition = self.parse_expression()312            self.reader.expect_token(")")313            statement = stats.DoStatement(condition, body)314        elif self.reader.read_token("for"):315            requires_closing = False316            self.reader.expect_token("(")317            var_decl_mod = self.reader.read_any_of(["const", "let", "var"])318            item_var_name = None if var_decl_mod == None else self.reader.expect_identifier()319            if item_var_name != None and self.reader.read_token("of"):320                items = self.parse_expression()321                self.reader.expect_token(")")322                body = self.expect_block_or_statement()323                statement = stats.ForeachStatement(stats.ForeachVariable(item_var_name), items, body)324            else:325                for_var = None326                if item_var_name != None:327                    type_and_init = self.parse_type_and_init()328                    for_var = stats.ForVariable(item_var_name, type_and_init.type, type_and_init.init)329                self.reader.expect_token(";")330                condition = self.parse_expression()331                self.reader.expect_token(";")332                incrementor = self.parse_expression()333                self.reader.expect_token(")")334                body = self.expect_block_or_statement()335                statement = stats.ForStatement(for_var, condition, incrementor, body)336        elif self.reader.read_token("try"):337            block = self.expect_block("try body is missing")338            339            catch_var = None340            catch_body = None341            if self.reader.read_token("catch"):342                self.reader.expect_token("(")343                catch_var = stats.CatchVariable(self.reader.expect_identifier(), None)344                self.reader.expect_token(")")345                catch_body = self.expect_block("catch body is missing")346            347            finally_body = self.expect_block() if self.reader.read_token("finally") else None348            return stats.TryStatement(block, catch_var, catch_body, finally_body)349        elif self.reader.read_token("return"):350            expr = None if self.reader.peek_token(";") else self.parse_expression()351            statement = stats.ReturnStatement(expr)352        elif self.reader.read_token("throw"):353            expr = self.parse_expression()354            statement = stats.ThrowStatement(expr)355        elif self.reader.read_token("break"):356            statement = stats.BreakStatement()357        elif self.reader.read_token("continue"):358            statement = stats.ContinueStatement()359        elif self.reader.read_token("debugger;"):360            return None361        else:362            expr = self.parse_expression()363            statement = stats.ExpressionStatement(expr)364            is_binary_set = isinstance(expr, exprs.BinaryExpression) and expr.operator in ["=", "+=", "-="]365            is_unary_set = isinstance(expr, exprs.UnaryExpression) and expr.operator in ["++", "--"]366            if not (isinstance(expr, exprs.UnresolvedCallExpression) or is_binary_set or is_unary_set or isinstance(expr, exprs.AwaitExpression)):367                self.reader.fail("this expression is not allowed as statement")368        369        if statement == None:370            self.reader.fail("unknown statement")371        372        statement.leading_trivia = leading_trivia373        self.node_manager.add_node(statement, start_pos)374        375        statement_last_line = self.reader.ws_line_counter376        if not self.reader.read_token(";") and requires_closing and self.reader.ws_line_counter == statement_last_line:377            self.reader.fail("statement is not closed", self.reader.ws_offset)378        379        return statement380    381    def parse_block(self):382        if not self.reader.read_token("{"):383            return None384        start_pos = self.reader.prev_token_offset385        386        statements = []387        if not self.reader.read_token("}"):388            while True:389                statement = self.expect_statement()390                if statement != None:391                    statements.append(statement)392                if not (not self.reader.read_token("}")):393                    break394        395        block = stats.Block(statements)396        self.node_manager.add_node(block, start_pos)397        return block398    399    def expect_block(self, error_msg = None):400        block = self.parse_block()401        if block == None:402            self.reader.fail(error_msg or "expected block here")403        return block404    405    def parse_type_args(self):406        type_arguments = []407        if self.reader.read_token("<"):408            while True:409                generics = self.parse_type()410                type_arguments.append(generics)411                if not (self.reader.read_token(",")):412                    break413            self.reader.expect_token(">")414        return type_arguments415    416    def parse_generics_args(self):417        type_arguments = []418        if self.reader.read_token("<"):419            while True:420                generics = self.reader.expect_identifier()421                type_arguments.append(generics)422                if not (self.reader.read_token(",")):423                    break424            self.reader.expect_token(">")425        return type_arguments426    427    def parse_expr_stmt_from_string(self, expression):428        expr = self.create_expression_parser(read.Reader(expression)).parse()429        return stats.ExpressionStatement(expr)430    431    def parse_method_signature(self, is_constructor, declaration_only):432        params = []433        fields = []434        if not self.reader.read_token(")"):435            while True:436                leading_trivia = self.reader.read_leading_trivia()437                param_start = self.reader.offset438                is_public = self.reader.read_token("public")439                if is_public and not is_constructor:440                    self.reader.fail("public modifier is only allowed in constructor definition")441                442                param_name = self.reader.expect_identifier()443                self.context.append(f'''arg:{param_name}''')444                type_and_init = self.parse_type_and_init()445                param = types.MethodParameter(param_name, type_and_init.type, type_and_init.init, leading_trivia)446                params.append(param)447                448                # init should be used as only the constructor's method parameter, but not again as a field initializer too449                #   (otherwise it would called twice if cloned or cause AST error is just referenced from two separate places)450                if is_public:451                    field = types.Field(param_name, type_and_init.type, None, types.VISIBILITY.PUBLIC, False, param, param.leading_trivia)452                    fields.append(field)453                    param.field_decl = field454                455                self.node_manager.add_node(param, param_start)456                self.context.pop()457                if not (self.reader.read_token(",")):458                    break459            460            self.reader.expect_token(")")461        462        returns = None463        if not is_constructor:464            # in case of constructor, "returns" won't be used465            returns = self.parse_type() if self.reader.read_token(":") else astTypes.VoidType.instance if self.missing_return_type_is_void else None466        467        body = None468        super_call_args = None469        if declaration_only:470            self.reader.expect_token(";")471        else:472            body = self.expect_block("method body is missing")473            first_stmt = body.statements[0] if len(body.statements) > 0 else None474            if isinstance(first_stmt, stats.ExpressionStatement) and isinstance(first_stmt.expression, exprs.UnresolvedCallExpression) and isinstance(first_stmt.expression.func, exprs.Identifier) and first_stmt.expression.func.text == "super":475                super_call_args = first_stmt.expression.args476                body.statements.pop(0)477        478        return MethodSignature(params, fields, body, returns, super_call_args)479    480    def parse_identifier_or_string(self):481        return self.reader.read_string() or self.reader.expect_identifier()482    483    def parse_interface(self, leading_trivia, is_exported):484        if not self.reader.read_token("interface"):485            return None486        intf_start = self.reader.prev_token_offset487        488        intf_name = self.reader.expect_identifier("expected identifier after 'interface' keyword")489        self.context.append(f'''I:{intf_name}''')490        491        intf_type_args = self.parse_generics_args()492        493        base_interfaces = []494        if self.reader.read_token("extends"):495            while True:496                base_interfaces.append(self.parse_type())497                if not (self.reader.read_token(",")):498                    break499        500        methods = []501        fields = []502        503        self.reader.expect_token("{")504        while not self.reader.read_token("}"):505            member_leading_trivia = self.reader.read_leading_trivia()506            507            member_start = self.reader.offset508            member_name = self.parse_identifier_or_string()509            if self.reader.read_token(":"):510                self.context.append(f'''F:{member_name}''')511                512                field_type = self.parse_type()513                self.reader.expect_token(";")514                515                field = types.Field(member_name, field_type, None, types.VISIBILITY.PUBLIC, False, None, member_leading_trivia)516                fields.append(field)517                518                self.node_manager.add_node(field, member_start)519                self.context.pop()520            else:521                self.context.append(f'''M:{member_name}''')522                method_type_args = self.parse_generics_args()523                self.reader.expect_token("(")524                # method525                   526                sig = self.parse_method_signature(False, True)527                528                method = types.Method(member_name, method_type_args, sig.params, sig.body, types.VISIBILITY.PUBLIC, False, sig.returns, False, member_leading_trivia)529                methods.append(method)530                self.node_manager.add_node(method, member_start)531                self.context.pop()532        533        intf = types.Interface(intf_name, intf_type_args, base_interfaces, fields, methods, is_exported, leading_trivia)534        self.node_manager.add_node(intf, intf_start)535        self.context.pop()536        return intf537    538    def parse_specified_type(self):539        type_name = self.reader.read_identifier()540        type_args = self.parse_type_args()541        return astTypes.UnresolvedType(type_name, type_args)542    543    def parse_class(self, leading_trivia, is_exported, declaration_only):544        cls_modifiers = self.reader.read_modifiers(["abstract"])545        if not self.reader.read_token("class"):546            return None547        cls_start = self.reader.prev_token_offset548        549        cls_name = self.reader.expect_identifier("expected identifier after 'class' keyword")550        self.context.append(f'''C:{cls_name}''')551        552        type_args = self.parse_generics_args()553        base_class = self.parse_specified_type() if self.reader.read_token("extends") else None554        555        base_interfaces = []556        if self.reader.read_token("implements"):557            while True:558                base_interfaces.append(self.parse_specified_type())559                if not (self.reader.read_token(",")):560                    break561        562        constructor = None563        fields = []564        methods = []565        properties = []566        567        self.reader.expect_token("{")568        while not self.reader.read_token("}"):569            member_leading_trivia = self.reader.read_leading_trivia()570            571            member_start = self.reader.offset572            modifiers = self.reader.read_modifiers(["static", "public", "protected", "private", "readonly", "async", "abstract"])573            is_static = "static" in modifiers574            is_async = "async" in modifiers575            is_abstract = "abstract" in modifiers576            visibility = types.VISIBILITY.PRIVATE if "private" in modifiers else types.VISIBILITY.PROTECTED if "protected" in modifiers else types.VISIBILITY.PUBLIC577            578            member_name = self.parse_identifier_or_string()579            method_type_args = self.parse_generics_args()580            if self.reader.read_token("("):581                # method582                is_constructor = member_name == "constructor"583                584                585                sig = self.parse_method_signature(is_constructor, declaration_only or is_abstract)586                if is_constructor:587                    member = constructor = types.Constructor(sig.params, sig.body, sig.super_call_args, member_leading_trivia)588                    for field in sig.fields:589                        fields.append(field)590                else:591                    method = types.Method(member_name, method_type_args, sig.params, sig.body, visibility, is_static, sig.returns, is_async, member_leading_trivia)592                    methods.append(method)593                    member = method594                595                self.node_manager.add_node(member, member_start)596            elif member_name == "get" or member_name == "set":597                # property598                prop_name = self.reader.expect_identifier()599                prop = next(filter(lambda x: x.name == prop_name, properties), None)600                prop_type = None601                getter = None602                setter = None603                604                if member_name == "get":605                    # get propName(): propType { return ... }606                    self.context.append(f'''P[G]:{prop_name}''')607                    self.reader.expect_token("()", "expected '()' after property getter name")608                    prop_type = self.parse_type() if self.reader.read_token(":") else None609                    if declaration_only:610                        if prop_type == None:611                            self.reader.fail("Type is missing for property in declare class")612                        self.reader.expect_token(";")613                    else:614                        getter = self.expect_block("property getter body is missing")615                        if prop != None:616                            prop.getter = getter617                elif member_name == "set":618                    # set propName(value: propType) { ... }619                    self.context.append(f'''P[S]:{prop_name}''')620                    self.reader.expect_token("(", "expected '(' after property setter name")621                    self.reader.expect_identifier()622                    prop_type = self.parse_type() if self.reader.read_token(":") else None623                    self.reader.expect_token(")")624                    if declaration_only:625                        if prop_type == None:626                            self.reader.fail("Type is missing for property in declare class")627                        self.reader.expect_token(";")628                    else:629                        setter = self.expect_block("property setter body is missing")630                        if prop != None:631                            prop.setter = setter632                633                if prop == None:634                    prop = types.Property(prop_name, prop_type, getter, setter, visibility, is_static, member_leading_trivia)635                    properties.append(prop)636                    self.node_manager.add_node(prop, member_start)637                638                self.context.pop()639            else:640                self.context.append(f'''F:{member_name}''')641                642                type_and_init = self.parse_type_and_init()643                self.reader.expect_token(";")644                645                field = types.Field(member_name, type_and_init.type, type_and_init.init, visibility, is_static, None, member_leading_trivia)646                fields.append(field)647                648                self.node_manager.add_node(field, member_start)649                self.context.pop()650        651        cls_ = types.Class(cls_name, type_args, base_class, base_interfaces, fields, properties, constructor, methods, is_exported, leading_trivia)652        self.node_manager.add_node(cls_, cls_start)653        self.context.pop()654        return cls_655    656    def parse_enum(self, leading_trivia, is_exported):657        if not self.reader.read_token("enum"):658            return None659        enum_start = self.reader.prev_token_offset660        661        name = self.reader.expect_identifier("expected identifier after 'enum' keyword")662        self.context.append(f'''E:{name}''')663        664        members = []665        666        self.reader.expect_token("{")667        if not self.reader.read_token("}"):668            while True:669                if self.reader.peek_token("}"):670                    break671                # eg. "enum { A, B, }" (but multiline)672                673                enum_member = types.EnumMember(self.reader.expect_identifier())674                members.append(enum_member)675                self.node_manager.add_node(enum_member, self.reader.prev_token_offset)676                677                # TODO: generated code compatibility678                self.reader.read_token(f'''= "{enum_member.name}"''')679                if not (self.reader.read_token(",")):680                    break681            self.reader.expect_token("}")682        683        enum_obj = types.Enum(name, members, is_exported, leading_trivia)684        self.node_manager.add_node(enum_obj, enum_start)685        self.context.pop()686        return enum_obj687    688    @classmethod689    def calculate_relative_path(cls, curr_file, rel_path):690        if not rel_path.startswith("."):691            raise Error(f'''relPath must start with \'.\', but got \'{rel_path}\'''')692        693        curr = re.split("/", curr_file)694        curr.pop()695        # filename does not matter696        for part in re.split("/", rel_path):697            if part == "":698                raise Error(f'''relPath should not contain multiple \'/\' next to each other (relPath=\'{rel_path}\')''')699            if part == ".":700                # "./" == stay in current directory701                continue702            elif part == "..":703                # "../" == parent directory704                if len(curr) == 0:705                    raise Error(f'''relPath goes out of root (curr=\'{curr_file}\', relPath=\'{rel_path}\')''')706                curr.pop()707            else:708                curr.append(part)709        return "/".join(curr)710    711    @classmethod712    def calculate_import_scope(cls, curr_scope, import_file):713        if import_file.startswith("."):714            # relative715            return types.ExportScopeRef(curr_scope.package_name, TypeScriptParser2.calculate_relative_path(curr_scope.scope_name, import_file))716        else:717            path = re.split("/", import_file)718            pkg_name = path.pop(0)719            return types.ExportScopeRef(pkg_name, types.Package.index if len(path) == 0 else "/".join(path))720    721    def read_identifier(self):722        raw_id = self.reader.read_identifier()723        return re.sub("_+$", "", raw_id)724    725    def parse_import(self, leading_trivia):726        if not self.reader.read_token("import"):727            return None728        import_start = self.reader.prev_token_offset729        730        import_all_alias = None731        import_parts = []732        733        if self.reader.read_token("*"):734            self.reader.expect_token("as")735            import_all_alias = self.reader.expect_identifier()736        else:737            self.reader.expect_token("{")738            while True:739                if self.reader.peek_token("}"):740                    break741                742                imp = self.reader.expect_identifier()743                if self.reader.read_token("as"):744                    self.reader.fail("This is not yet supported")745                import_parts.append(types.UnresolvedImport(imp))746                self.node_manager.add_node(imp, self.reader.prev_token_offset)747                if not (self.reader.read_token(",")):748                    break749            self.reader.expect_token("}")750        751        self.reader.expect_token("from")752        module_name = self.reader.expect_string()753        self.reader.expect_token(";")754        755        import_scope = TypeScriptParser2.calculate_import_scope(self.export_scope, module_name) if self.export_scope != None else None756        757        imports = []758        if len(import_parts) > 0:759            imports.append(types.Import(import_scope, False, import_parts, None, leading_trivia))760        761        if import_all_alias != None:762            imports.append(types.Import(import_scope, True, None, import_all_alias, leading_trivia))763        #this.nodeManager.addNode(imports, importStart);764        return imports765    766    def parse_source_file(self):767        imports = []768        enums = []769        intfs = []770        classes = []771        funcs = []772        while True:773            leading_trivia = self.reader.read_leading_trivia()774            if self.reader.get_eof():775                break776            777            imps = self.parse_import(leading_trivia)778            if imps != None:779                for imp in imps:780                    imports.append(imp)781                continue782            783            modifiers = self.reader.read_modifiers(["export", "declare"])784            is_exported = "export" in modifiers785            is_declaration = "declare" in modifiers786            787            cls_ = self.parse_class(leading_trivia, is_exported, is_declaration)788            if cls_ != None:789                classes.append(cls_)790                continue791            792            enum_obj = self.parse_enum(leading_trivia, is_exported)793            if enum_obj != None:794                enums.append(enum_obj)795                continue796            797            intf = self.parse_interface(leading_trivia, is_exported)798            if intf != None:799                intfs.append(intf)800                continue801            802            if self.reader.read_token("function"):803                func_name = self.read_identifier()804                self.reader.expect_token("(")805                sig = self.parse_method_signature(False, is_declaration)806                funcs.append(types.GlobalFunction(func_name, sig.params, sig.body, sig.returns, is_exported, leading_trivia))807                continue808            809            break810        811        self.reader.skip_whitespace()812        813        stmts = []814        while True:815            leading_trivia = self.reader.read_leading_trivia()816            if self.reader.get_eof():...

libgen.py

Source:libgen.py

...72    def __init__ (self, file):73        self.f = file74        self.current_line = 075        self._didremovecomments = False76    def read_token(self):77        self.f.tell()  # <--- EXTREMELY IMPORTANT, do not remove or bugs will appear on parsing78        line = self.f.readline()79        #print("pos", hex(curr), "Line", line)80        #print(hex(self.f.tell()))81        self.current_line += 182        if not line:83            return ""84        line = line.strip()85        comment = line.find("#")86        if comment != -1:87            line = line[:comment]88            self._didremovecomments = True89        comment2 = line.find("//")90        if comment2 != -1:91            line = line[:comment2]92            self._didremovecomments = True93        if line.strip() == "":94            line = self.read_token()  # Try reading the next token95        return line.strip()96    def peek_token(self):97        #print("next is a peek")98        curr = self.f.tell()99        currline = self.current_line100        next_token = self.read_token()101        self.current_line = currline102        self.f.seek(curr)103        #print(curr, self.f.tell())104        assert curr == self.f.tell()105        #print("peek done, result:", next_token)106        return next_token107    def read_section_rest_raw(self):108        curr = f.tell()109        self.skip_current_section()110        end = f.tell()111        f.seek(curr)112        rest = f.read(end-curr)113        f.seek(curr)114        return rest115    def skip_next_section(self):116        token = self.read_token()117        if token == "{":118            level = 1119            while level != 0:120                token = self.read_token()121                if token == "{":122                    level += 1123                elif token == "}":124                    level -= 1125                elif token == "":126                    raise RuntimeError("Reached end of file while skipping {{ }} block. File is likely malformed")127        else:128            raise RuntimeError("Expected '{{' for start of section, instead got {0}".format(token))129    def skip_current_section(self):130        level = 0131        while level != -1:132            token = self.read_token()133            if token == "{":134                level += 1135            elif token == "}":136                level -= 1137            elif token == "":138                raise RuntimeError("Reached end of file while skipping to end of current { } block. File is likely malformed.")139    def read_vector3f(self):140        val = self.read_token()141        floats = val.split(" ")142        if len(floats) != 3:143            raise RuntimeError("Tried to read Vector3f but got {0}".format(floats))144        return float(floats[0]), float(floats[1]), float(floats[2])145    def read_integer(self):146        val = self.read_token()147        if val == "":148            raise RuntimeError("Reached end of file while reading integer!")149        return int(val)150    def read_float(self):151        val = self.read_token()152        if val == "":153            raise RuntimeError("Reached end of file while reading float!")154        return float(val)155    def read_string(self):156        val = self.read_token()157        #print(val)158        #assert val[0] == "\"" and val[-1] == "\""159        syntax_assert(val[0] == "\"" and val[-1] == "\"",160                      "Malformed String", self.current_line, val)161        return val[1:-1]162    def read_string_tripple(self):163        val = self.read_token()164        tripple = []165        start = None166        #for i in range(3):167        for i, char in enumerate(val):168            if char == "\"" and start is None:169                start = i170            elif char == "\"" and start is not None:171                tripple.append(val[start:i+1])172                start = None173        if start is not None:174            raise RuntimeError("Malformed string tripple {0}".format(val))175        return tripple176    def read_float_int(self):177        val = self.read_token()178        f, i = val.split(" ")179        return float(f), int(i)180    def read_int_string(self):181        val = self.read_token()182        i, s = val.split(" ")183        s = s.strip()184        syntax_assert(s[0] == "\"" and s[-1] == "\"",185                      "Malformed String", self.current_line, val)186        return int(i), s[1:-1]187class GeneratorParameters(object):188    pass189class GeneratorObject(object):190    def __init__(self, name, version, generatorid=["", "", ""], modes=[1, 1, 1], fid=0, fmt=0):191        self.name = name192        self.version = version193        self.generatorid = generatorid194        self.modes = modes195        self.fid = fid196        self.fmt = fmt197        self.spline = []198        self.spline_float = None199        self.spline_params = []200        self.position = Vector3(0, 0, 0)201        self.rotation = Vector3(0, 0, 0)202        self.scale = 1.0203        self.unknown_params = OrderedDict()204    def from_other(self, obj):205        self.name = obj.name206        self.version = obj.version207        self.generatorid = obj.generatorid208        self.spline = obj.spline209        self.spline_params = obj.spline_params210        self.position = obj.position211        self.rotation = obj.rotation212        self.scale = obj.scale213        self.unknown_params = obj.unknown_params214    def copy(self):215        return deepcopy(self)216    @classmethod217    def from_generator_file(cls, reader: GeneratorReader):218        name = reader.read_string()219        #print("NOW WE ARE DOING ", name)220        version = reader.read_string()221        generatorid = reader.read_string_tripple()222        if int(version) >= 6:223            print("version 6 or higher, getting mode booleans")224            modes = [225                reader.read_token(),226                reader.read_token(),227                reader.read_token()228            ]229            if int(version) >= 7:230                print("version 7 or higher, getting fid and fmt")231                fid = reader.read_token()232                fmt = reader.read_token()233                gen = cls(name, version, generatorid, modes, fid, fmt)234            else:235                gen = cls(name, version, generatorid, modes)236        else:237            print("version lower than 6")238            gen = cls(name, version, generatorid)239        gen.read_parameters(reader)240        gen._read_spline(reader)241        return gen242    def write(self, writer: GeneratorWriter):243        writer.write_string(self.name)244        writer.write_string(self.version)245        writer.write_string_tripple(*self.generatorid)246        if int(self.version) >= 6:247            writer.write_token(self.modes[0], "# Normal [0=false, 1=true]")248            writer.write_token(self.modes[1], "# Hard [0=false, 1=true]")249            writer.write_token(self.modes[2], "# Ultra-Spicy [0=false, 1=true]")250            if int(self.version) >= 7:251                writer.write_token(self.fid, "# Face ID (Unknown)")252                writer.write_token(self.fmt, "# Face Message Table (Unkown)")253        254        self.write_parameters(writer)255        if len(self.spline) == 0:256            writer.write_string("no-spline")257        else:258            writer.write_string("spline")259            writer.write_integer(len(self.spline))260            for x, y, z in self.spline:261                writer.write_vector3f(x, y, z)262            writer.write_float_int(self.spline_float, len(self.spline_params))263            for id, name, params in self.spline_params:264                writer.write_int_string(id, name)265                writer.open_bracket()266                for paramname, paramval in params.items():267                    writer.open_bracket()268                    writer.write_string(paramname)269                    writer.write_token(paramval)270                    writer.close_bracket()271                writer.close_bracket()272    def _get_comments(self, param):273        if param == "mSaveOption":274            return ["flag", "birthday", "deadday", "interval", "expireday", "curr num", "expire progress"]275        else:276            return []277    def write_parameters(self, writer:GeneratorWriter):278        writer.open_bracket()279        writer.open_bracket()280        writer.write_string("mPos")281        writer.write_vector3f(self.position.x, self.position.y, self.position.z)282        writer.close_bracket()283        writer.open_bracket()284        writer.write_string("mBaseScale")285        writer.write_float(self.scale)286        writer.close_bracket()287        writer.open_bracket()288        writer.write_string("mPosture")289        writer.write_vector3f(self.rotation.x, self.rotation.y, self.rotation.z)290        writer.close_bracket()291        for param, values in self.unknown_params.items():292            comments = self._get_comments(param)293            writer.open_bracket()294            writer.write_string(param)295            if param == "mEmitRadius":296                writer.write_float(values)297            else:298                level = 0299                for i, val in enumerate(values):300                    if val == "{":301                        writer.open_bracket()302                        level += 1303                    elif val == "}":304                        writer.close_bracket()305                        level -= 1306                    else:307                        if i < len(comments):308                            writer.write_token(val, "# "+comments[i])309                        else:310                            writer.write_token(val)311                syntax_assert(level==0, "Bracket mismatch", writer.current_line)312            writer.close_bracket()313        writer.close_bracket()314    def read_parameters(self, reader: GeneratorReader):315        if reader.read_token() != "{":316            raise RuntimeError("")317        next = reader.read_token()318        if next == "":319            raise RuntimeError("Tried to read parameters but encountered EOF")320        assert next in ("{", "}")321        while next != "}":322            param_name = reader.read_string()323            if param_name == "mPos":324                self.position = Vector3(*reader.read_vector3f())325                reader.read_token()326            elif param_name == "mPosture":327                self.rotation = Vector3(*reader.read_vector3f())328                reader.read_token()329            elif param_name == "mBaseScale":330                self.scale = reader.read_float()331                reader.read_token()332            elif param_name == "mEmitRadius":333                self.unknown_params[param_name] = reader.read_float()334                reader.read_token()335            else:336                unkdata = []337                level = 0338                while level != -1:339                    subnext = reader.read_token()340                    if subnext == "":341                        raise RuntimeError("Encountered EOF while reading parameter")342                    elif subnext == "{":343                        level += 1344                    elif subnext == "}":345                        level -= 1346                    if level != -1:347                        unkdata.append(subnext)348                self.unknown_params[param_name] = unkdata349            next = reader.read_token()350            syntax_assert(next != "",351                          "Reached end of file while parsing parameters",352                          reader.current_line)353            syntax_assert(next in ("{", "}"),354                          "Malformed file, expected {{ or }} but got {0}".format(next),355                          reader.current_line)356            #assert next in ("{", "}", "")357    def _read_spline(self, reader: GeneratorReader):358        splinetext = reader.read_string()359        if splinetext == "no_spline":360            pass361        elif splinetext == "spline":362            spline_count = reader.read_integer()363            for i in range(spline_count):364                self.spline.append(reader.read_vector3f())365            self.spline_float, paramcount = reader.read_float_int()366            self.spline_params = []367            for i in range(paramcount):368                id, splinename = reader.read_int_string()369                assert reader.read_token() == "{"370                next = reader.read_token()371                assert next != ""372                params = OrderedDict()373                while next != "}":374                    assert next == "{"375                    paramname = reader.read_string()376                    paramval = reader.read_token()377                    assert reader.read_token() == "}"378                    next = reader.read_token()379                    params[paramname] = paramval380                self.spline_params.append((id, splinename, params))381class GeneratorFile(object):382    def __init__(self):383        self.generators = []384    @classmethod385    def from_file(cls, f):386        """data = f.read()387        if "#" not in data:388            #print(data)389            #assert "#" in f.read()390            dontsave = True391        else:392            dontsave = False393        f.seek(0)"""394        genfile = cls()395        reader = GeneratorReader(f)396        written = {}397        try:398            start = reader.read_token()399            if start != "{":400                raise RuntimeError("Expected file to start with '{'")401            next = reader.peek_token()402            if next == "":403                raise RuntimeError("Malformed file, expected generator object or '}'")404            while next != "}":405                start = reader.f.tell()406                generator = GeneratorObject.from_generator_file(reader)407                end = reader.f.tell()408                #print(generator.name)409                genfile.generators.append(generator)410                next = reader.peek_token()411                #assert reader.peek_token() == next412                #assert reader.peek_token() == next...

test_client.py

Source:test_client.py

1#!/usr/bin/env python2# -*- coding: utf-8 -*-3import uuid4from mock import mock5import requests6import warp10client7from warp10client.tests import base8from warp10client.timeserie import Timeserie9class TestWarp10ClientTestCase(base.BaseTestCase):10    def setUp(self):11        super(TestWarp10ClientTestCase, self).setUp()12        self.mock_session = mock.Mock()13        self.mock_response = mock.Mock()14        self.write_token = str(uuid.uuid4())15        self.read_token = str(uuid.uuid4())16        self.resource_id = str(uuid.uuid4())17        self.project_id = str(uuid.uuid4())18        self.warp10_url = 'http://example.warp10.com'19        self.metric_name = 'cpu_util'20        self.app = 'example.project_name'21        self.expected_value = 0.72190925525522        self.expected_timestamp = 150639840000000023        self.expected_unit = '%'24        self.metric_write = {25            'name': self.metric_name,26            'tags': {27                'resource_id': self.resource_id,28                'project_id': self.project_id,29                'unit': self.expected_unit,30            },31            'position': {32                'longitude': None,33                'latitude': None,34                'elevation': None,35                'timestamp': self.expected_timestamp,36            },37            'value': self.expected_value,38        }39        self.metric_get = {40            'name': self.metric_name,41            'tags': {42                'resource_id': self.resource_id,43                'project_id': self.project_id,44            },45            'aggregate': {46                'type': 'mean',47                'span': 1000000 * 3600,48            },49            'timestamp': {50                'start': "2017-01-01T00:00:00.000Z",51                'end': "2018-01-01T00:00:00.000Z"52            }53        }54    def test_call_fetch(self):55        self.mock_response.status_code = 20056        self.mock_session.post = mock.Mock(return_value=self.mock_response)57        expected_url = self.warp10_url + '/exec'58        expected_headers = {59            'X-Warp10-Token': self.read_token,60        }61        with mock.patch('requests.Session', return_value=self.mock_session):62            client = warp10client.Warp10Client(write_token=self.write_token,63                                               read_token=self.read_token,64                                               warp10_api_url=self.warp10_url)65            client._call(self.metric_get, call_type='fetch')66            self.mock_session.post.assert_called_with(expected_url,67                                                      data=mock.ANY,68                                                      headers=expected_headers)69    def test_call_ingress(self):70        self.mock_response.status_code = 20071        self.mock_session.post = mock.Mock(return_value=self.mock_response)72        expected_url = self.warp10_url + '/update'73        expected_headers = {74            'X-Warp10-Token': self.write_token,75            'Content-Type': 'text/plain'76        }77        with mock.patch('requests.Session', return_value=self.mock_session):78            client = warp10client.Warp10Client(write_token=self.write_token,79                                               read_token=self.read_token,80                                               warp10_api_url=self.warp10_url)81            client._call(self.metric_write, call_type='ingress')82            self.mock_session.post.assert_called_with(expected_url,83                                                      data=mock.ANY,84                                                      headers=expected_headers)85    def test_get_resp_200(self):86        self.mock_response.status_code = 20087        self.mock_session.post = mock.Mock(return_value=self.mock_response)88        string_args = [self.metric_name, self.resource_id,89                       self.project_id, self.app, "{}",90                       self.expected_timestamp, self.expected_value]91        resp_content = '[[{{"c":"{}","l":{{"resource_id":"{}",'\92                       '"project_id":"{}",".app":"{}"}},"a":{}'\93                       ',"v":[[{},{}]]}}]]'.format(*string_args)94        self.mock_response.content = resp_content95        expected_tags = {96            'resource_id': str(self.resource_id),97            'project_id': str(self.project_id),98            '.app': self.app99        }100        with mock.patch('requests.Session', return_value=self.mock_session):101            client = warp10client.Warp10Client(write_token=self.write_token,102                                               read_token=self.read_token,103                                               warp10_api_url=self.warp10_url)104            timeserie = client.get(self.metric_get)105            self.assertIsInstance(timeserie, Timeserie)106            self.assertEqual(timeserie.metrics[0].name, self.metric_name)107            self.assertEqual(timeserie.metrics[0].value, self.expected_value)108            self.assertEqual(timeserie.metrics[0]._tags, expected_tags)109            self.assertEqual(timeserie.metrics[0].position.timestamp,110                             self.expected_timestamp)111    def test_get_resp_503(self):112        self.mock_response.status_code = 503113        self.mock_session.post = mock.Mock(return_value=self.mock_response)114        with mock.patch('requests.Session', return_value=self.mock_session):115            client = warp10client.Warp10Client(write_token=self.write_token,116                                               read_token=self.read_token,117                                               warp10_api_url=self.warp10_url)118            self.assertRaises(requests.RequestException,119                              client.get, self.metric_get)120    def test_exists_metric_exists(self):121        mock_response = mock.Mock()122        mock_response.status_code.return_value = 200123        string_args = [self.metric_name, self.resource_id,124                       self.project_id, self.app, "{}",125                       self.expected_timestamp, self.expected_value]126        resp_content = '[[{{"c":"{}","l":{{"resource_id":"{}",'\127                       '"project_id":"{}",".app":"{}"}},"a":{}'\128                       ',"v":[[{},{}]]}}]]'.format(*string_args)129        mock_response.content = resp_content130        with mock.patch(131                'requests.Session', return_value=self.mock_session132        ), mock.patch(133                'warp10client.client.Warp10Client._call',134                return_value=mock_response135        ):136            client = warp10client.Warp10Client(write_token=self.write_token,137                                               read_token=self.read_token,138                                               warp10_api_url=self.warp10_url)139            self.assertTrue(client.exists(self.metric_get))140    def test_exists_non_exist(self):141        mock_response = mock.Mock()142        mock_response.status_code.return_value = 200143        resp_content = '[[]]'144        mock_response.content = resp_content145        with mock.patch(146                'requests.Session', return_value=self.mock_session147        ), mock.patch(148                'warp10client.client.Warp10Client._call',149                return_value=mock_response150        ):151            client = warp10client.Warp10Client(write_token=self.write_token,152                                               read_token=self.read_token,153                                               warp10_api_url=self.warp10_url)154            self.assertFalse(client.exists(self.metric_get))155    def test_delete(self):156        with mock.patch('requests.Session', return_value=self.mock_session):157            client = warp10client.Warp10Client(write_token=self.write_token,158                                               read_token=self.read_token,159                                               warp10_api_url=self.warp10_url)160            self.assertRaises(NotImplementedError,161                              client.delete, self.metric_write)162    def test__call__gen_requests_body(self):163        with mock.patch('warp10client.client.Warp10Client._gen_request_body',164                        side_effect=warp10client.client.CallException):165            client = warp10client.Warp10Client(write_token=self.write_token,166                                               read_token=self.read_token,167                                               warp10_api_url=self.warp10_url)168            self.assertRaises(warp10client.client.CallException,169                              client._call, self.metric_write)170    def test__call__get_method(self):171        with mock.patch('warp10client.client.Warp10Client._get_method',172                        side_effect=warp10client.client.CallException):173            client = warp10client.Warp10Client(write_token=self.write_token,174                                               read_token=self.read_token,175                                               warp10_api_url=self.warp10_url)176            self.assertRaises(warp10client.client.CallException,...

pdf_parser.py

Source:pdf_parser.py

...17		index = index + 118def view_token() :19	global index20	save_index = index21	t = read_token()22	index = save_index23	return t24def read_token() :25	global index26	skip_blanks()27	token = ''28	c = contents[index]29	if is_special(c) :30		if contents[index:index+2] in [ '<<', '>>' ] :31			token = contents[index:index+2]32			index = index + 233		else :34			token = c35			index = index + 136	else :37		while not is_blank(c) and not is_special(c) :38			token = token + c39			index = index + 140			c = contents[index]41	return token42def read_dictionnary() :43	d = { }44	while view_token() != '>>' :45		name = read_element() # should be /name46		element = read_element()47		#print "++", "element=", element48		try :49			d[name] = element50		except :51			print 'name=', name52			raise53	54	read_token() # consume '>>'55	return d56def read_string() :57	global index58	level = 059	s = ''60	c = contents[index]61	while level >= 0 :62		if c == ')' : level = level - 163		if c == '(' : level = level + 164		if level >=0 : s = s + c65		index = index + 166		c = contents[index]67	68	index = index + 1 # consume ')'69	return s70def read_array() :71	a = [ ]72	while view_token() != ']' :73		element = read_element()74		a.append(element)75	76	read_token() # consume ']'77	return a78	79	80def read_element() :81	global index82	t = read_token()83	if t == '<<' : element = read_dictionnary()84	elif t == '/' : element = '/' + read_token()85	#elif t == '<' : element = read_hexstring()86	elif t == '[' : element = read_array()87	elif t == '(' : element = read_string()88	else :89		if t.isdigit() : # find out if is it a reference90			save_index = index91			a = read_token()92			r = read_token()93			if r == 'R' : # it is a reference94				element = [ 'R', t, a ]95			else :96				index = save_index97				element = t98		else :99			element = t100	return element101	102def skip_to_endstream() :103	global index104	while contents[index:index+9] != 'endstream' :105		index = index + 1106	index = index + 9107def read_obj(id) :108	id = int(id)109	id_gen = int(read_token())110	t = read_token()111	if t != 'obj' :112		# error113		raise Exception('obj expected ! (found %s)' % t)114	read_element()115	t = read_token()116	if t == 'stream' :117		#print 'stream'118		skip_to_endstream()119		read_token() # should be endobj120		#print 'endobj'121	elif t == 'endobj' :122		#print 'endobj'123		None124	else :125		raise Exception('not supported %s in read_obj()' % t)126	127def read_startxref() :128	t = read_token() # should be 'startxref'129	t = read_token()130def read_trailer() :131	t = read_token() # should be '<<'132	print 'read_trailer: t=', t133	read_dictionnary()134	read_startxref()135def read_xref() :136	while read_token() != 'trailer' :137		None138	read_trailer()139while True :140	t = read_token()141	if t == 'xref' :142		read_xref()143	elif t == '%' :144		if contents[index:index+4] == '%EOF' : break145		print 'comment'146		skip_to_eol()147	else :148		# obj149#		print 'tok=', t...

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.