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

How to use is_collect_metrics_mode method in localstack

Best Python code snippet using localstack_python

conftest.py

Source:conftest.py

...23localstack_stopped = mp.Event()  # event indicating that localstack has been stopped24startup_monitor_event = mp.Event()  # event that can be triggered to start localstack25# collection of functions that should be executed to initialize tests26test_init_functions = set()27if config.is_collect_metrics_mode():28    pytest_plugins = "localstack.testing.pytest.metric_collection"29@pytest.hookimpl()30def pytest_configure(config):31    # first pytest lifecycle hook32    _start_monitor()33def pytest_runtestloop(session):34    # second pytest lifecycle hook (before test runner starts)35    # collect test classes36    test_classes = set()37    for item in session.items:38        if item.parent and item.parent.cls:39            test_classes.add(item.parent.cls)40        # OpenSearch/Elasticsearch are pytests, not unit test classes, so we check based on the item parent's name.41        # Any pytests that rely on opensearch/elasticsearch must be special-cased by adding them to the list below...

metric_handler.py

Source:metric_handler.py

...102        self.metrics_handler_items = {}103    def create_metric_handler_item(104        self, chain: HandlerChain, context: RequestContext, response: Response105    ):106        if not config.is_collect_metrics_mode():107            return108        item = MetricHandlerItem(context)109        self.metrics_handler_items[context] = item110    def _get_metric_handler_item_for_context(self, context: RequestContext) -> MetricHandlerItem:111        return self.metrics_handler_items[context]112    def record_parsed_request(113        self, chain: HandlerChain, context: RequestContext, response: Response114    ):115        if not config.is_collect_metrics_mode():116            return117        item = self._get_metric_handler_item_for_context(context)118        item.parameters_after_parse = (119            list(context.service_request.keys()) if context.service_request else []120        )121    def record_exception(122        self, chain: HandlerChain, exception: Exception, context: RequestContext, response: Response123    ):124        if not config.is_collect_metrics_mode():125            return126        item = self._get_metric_handler_item_for_context(context)127        item.caught_exception_name = exception.__class__.__name__128    def update_metric_collection(129        self, chain: HandlerChain, context: RequestContext, response: Response130    ):131        if not config.is_collect_metrics_mode() or not context.service_operation:132            return133        is_internal = is_internal_call_context(context.request.headers)134        item = self._get_metric_handler_item_for_context(context)135        # parameters might get changed when dispatched to the service - we use the params stored in136        # parameters_after_parse137        parameters = ",".join(item.parameters_after_parse or [])138        response_data = response.data.decode("utf-8") if response.status_code >= 300 else ""139        MetricHandler.metric_data.append(140            Metric(141                service=context.service_operation.service,142                operation=context.service_operation.operation,143                headers=context.request.headers,144                parameters=parameters,145                response_code=response.status_code,...

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.