Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use incr_stats method in locust

Best Python code snippet using locust

api.py

Source:api.py

...248              $ref: '#/components/responses/422'249            500:250              $ref: '#/components/responses/500'251        """252        self.incr_stats("init", self.table_metadata.__name__)253        try:254            table_info: Dict = get_table_metadata(database, table_name, schema_name)255        except SQLAlchemyError as e:256            self.incr_stats("error", self.table_metadata.__name__)257            return self.response_422(error_msg_from_exception(e))258        self.incr_stats("success", self.table_metadata.__name__)259        return self.response(200, **table_info)260    @expose("/<int:pk>/select_star/<string:table_name>/", methods=["GET"])261    @expose(262        "/<int:pk>/select_star/<string:table_name>/<string:schema_name>/",263        methods=["GET"],264    )265    @protect()266    @check_datasource_access267    @safe268    @event_logger.log_this269    def select_star(270        self, database: Database, table_name: str, schema_name: Optional[str] = None271    ):272        """ Table schema info273        ---274        get:275          description: Get database select star for table276          parameters:277          - in: path278            schema:279              type: integer280            name: pk281            description: The database id282          - in: path283            schema:284              type: string285            name: table_name286            description: Table name287          - in: path288            schema:289              type: string290            name: schema_name291            description: Table schema292          responses:293            200:294              description: select star for table295              content:296                text/plain:297                  schema:298                    type: object299                    properties:300                      result:301                        type: string302                        description: SQL select star303            400:304              $ref: '#/components/responses/400'305            401:306              $ref: '#/components/responses/401'307            404:308              $ref: '#/components/responses/404'309            422:310              $ref: '#/components/responses/422'311            500:312              $ref: '#/components/responses/500'313        """314        self.incr_stats("init", self.select_star.__name__)315        try:316            result = database.select_star(317                table_name, schema_name, latest_partition=True, show_cols=True318            )319        except NoSuchTableError:320            self.incr_stats("error", self.select_star.__name__)321            return self.response(404, message="Table not found on the database")322        self.incr_stats("success", self.select_star.__name__)...

listener.py

Source:listener.py

...50            '''51            #if (begreq.flags & FCGI_KEEP_CONN) == 0:52            if not keep_conn:53                if protocol.close_socket(conn):54                    context.incr_stats('socket-closed', 'response-ok')55                    print('socket closed', file=sys.stderr)56        except Exception as e:57            raise pyfcgi._SystemError(reqid) from e58    # end while True59@pyfcgi.report_exception60def send_http_error(conn:socket.socket, requestId:int, errmsg:str, exinfo:tuple, http_code:int):61    http_mesg = http.client.responses[http_code]62    headers = {63        pyfcgi.CONST_STATUS: f'{http_code} {http_mesg}',64        pyfcgi.CONST_CONTENT_TYPE: 'text/html; charset=iso-8859-1',65    }66    errcode = str(uuid.uuid4())67    textmsg = f'error-code={errcode}'68    htmlmsg = f'<html><body>{textmsg}</body></html>'69    hresp = pyfcgi.Response(headers, htmlmsg)70    pyfcgi.send_record(conn, recordType=protocol.FCGI_STDOUT, requestId=requestId, contentData=hresp)71    pyfcgi.send_record(conn, recordType=protocol.FCGI_STDOUT, requestId=requestId)72    logmsg = traceback.format_exception(*exinfo)73    logmsg = f'{textmsg}; {errmsg}; ' + '; '.join(( v.replace('\n', '').strip() for v in logmsg ))74    pyfcgi.send_record(conn, recordType=protocol.FCGI_STDERR, requestId=requestId, contentData=logmsg)75    pyfcgi.send_record(conn, recordType=protocol.FCGI_STDERR, requestId=requestId)76    # end77    endreq = protocol.FCGI_EndRequestBody(2, protocol.FCGI_REQUEST_COMPLETE)78    pyfcgi.send_record(conn, recordType=protocol.FCGI_END_REQUEST, requestId=requestId, contentData=endreq.dump())79def on_accepted(context:pyfcgi.Context, conn:socket.socket, client:tuple):80    exinfo = None81    try:82        do_response(context, conn, client)83    except pyfcgi._SystemError as e:84        exinfo = sys.exc_info()85        try:86            if not isinstance(e.__context__, ConnectionError):87                send_http_error(conn, e.requestId, str(e.__context__), exinfo, http.client.INTERNAL_SERVER_ERROR)88        except:89            # ignore90            pass91    except Exception as e:92        exinfo = sys.exc_info()93    if exinfo:94        traceback.print_exception(*exinfo, file=sys.stderr)95        if protocol.close_socket(conn):96            context.incr_stats('socket-closed', 'response-ng')97            print('socket closed', file=sys.stderr)98def accept_submit(context:pyfcgi.Context, executor, ssock:socket.socket):99    try:100        csock, address = ssock.accept()101        context.incr_stats('socket-accepted')102        if context.loop:103            context.handler(pyfcgi.Event('ACCEPT'))104            executor.submit(on_accepted, context, csock, address)105    except BlockingIOError as e:106        context.incr_stats('socket-blockerr')107        print(f'{os.getpid()=} {str(e)}, ignore', file=sys.stderr)108    except socket.timeout as e:109        context.incr_stats('socket-timeout')110        context.handler(pyfcgi.Event('IDLE'))111def nonblocking_loop(context:pyfcgi.Context, ssock:socket.socket):112    with concurrent.futures.ThreadPoolExecutor(max_workers=context.threads) as executor, \113         selectors.DefaultSelector() as selector:114        '''115        https://docs.python.org/ja/3/library/socket.html#socket-timeouts116        sock.setblocking(True) ã¯ sock.settimeout(None) ã¨çä¾¡ã§ã117        sock.setblocking(False) ã¯ sock.settimeout(0.0) ã¨çä¾¡ã§ã118        '''119        ssock.setblocking(False)120        # https://docs.python.org/ja/3/library/selectors.html121        a = functools.partial(accept_submit, context, executor)122        selector.register(ssock, selectors.EVENT_READ, a)123        while context.loop:124            context.incr_stats('nonblocking-loop')125            ready = selector.select(context.so_timeout)126            if ready:127                for key, _ in ready:128                    callback = key.data129                    callback(ssock)130            else:131                context.incr_stats('select-timeout')132                context.handler(pyfcgi.Event('IDLE'))133def blocking_loop(context:pyfcgi.Context, ssock:socket.socket):134    with concurrent.futures.ThreadPoolExecutor(max_workers=context.threads) as executor, \135         selectors.DefaultSelector() as selector:136        ssock.settimeout(context.so_timeout)137        while context.loop:138            context.incr_stats('bloking-loop')139            accept_submit(context, executor, ssock)140def unlink_bind_file(bind_path:str):141    if os.path.exists(bind_path):142        print(f'unlink {bind_path=}', file=sys.stderr)143        os.unlink(bind_path)144@pyfcgi.report_exception145def start(context:pyfcgi.Context):146    context.handler(pyfcgi.Event('START-LISTENER'))147    oldmask = None148    if type(context.bind_addr) == str:149        family = socket.AF_UNIX150        if os.path.exists(context.bind_addr):151            print(f'unlink {context.bind_addr=}', file=sys.stderr)152            os.unlink(context.bind_addr)...

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.