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

How to use statsd_client method in pytest-play

Best Python code snippet using pytest-play_python

test_stats.py

Source:test_stats.py

...75            assert isinstance(airflow.stats.Stats.statsd, statsd.StatsClient)76            assert not hasattr(airflow.stats.Stats, 'dogstatsd')77        # Avoid side-effects78        importlib.reload(airflow.stats)79    def test_load_custom_statsd_client(self):80        with conf_vars(81            {82                ("metrics", "statsd_on"): "True",83                ("metrics", "statsd_custom_client_path"): f"{__name__}.CustomStatsd",84            }85        ):86            importlib.reload(airflow.stats)87            assert isinstance(airflow.stats.Stats.statsd, CustomStatsd)88        # Avoid side-effects89        importlib.reload(airflow.stats)90    def test_load_invalid_custom_stats_client(self):91        with conf_vars(92            {93                ("metrics", "statsd_on"): "True",...

test_statsd_client.py

Source:test_statsd_client.py

...3import pytest4import aio_statsd5pytestmark = pytest.mark.asyncio6@pytest.fixture7async def statsd_client() -> AsyncGenerator[aio_statsd.StatsdClient, None]:8    client: aio_statsd.StatsdClient = aio_statsd.StatsdClient(port=9999)9    await client.connect()10    yield client11    await client.close()12class TestStatsdClient:13    async def test_counter(self, statsd_client: aio_statsd.StatsdClient, udp_server: asyncio.Queue) -> None:14        statsd_client.counter("test.key", 1)15        assert await udp_server.get() == b"test.key:1|c"16    async def test_gauge(self, statsd_client: aio_statsd.StatsdClient, udp_server: asyncio.Queue) -> None:17        statsd_client.gauge("test.key", 1)18        assert await udp_server.get() == b"test.key:1|g"19    async def test_sets(self, statsd_client: aio_statsd.StatsdClient, udp_server: asyncio.Queue) -> None:20        statsd_client.sets("test.key", 1)21        assert await udp_server.get() == b"test.key:1|s"...

statsd.py

Source:statsd.py

1# -*- coding: utf-8 -*-2from asyncio.tasks import Task3import structlog4from ..typedefs import HTTPRequest5from ..typedefs import HTTPResponse6from ..utils import async_nowait_middleware7from ..async_stats import fmt_timings8logger = structlog.get_logger(__name__)9# pylint: disable=no-member,pointless-statement,protected-access10async def init_stats(request: HTTPRequest) -> None:11    try:12        statsd_client = getattr(request.app.config, 'statsd_client', None)13        if not statsd_client:14            return15        if request.is_single_jrpc:16            statsd_client.incr('jrpc.inflight')17        elif request.is_batch_jrpc and statsd_client:18            _ = [statsd_client.incr('jrpc.inflight') for r in request.jsonrpc]19    except BaseException as e:20        logger.warning('send_stats', e=e)21# pylint: disable=unused-argument22@async_nowait_middleware23async def send_stats(request: HTTPRequest,24                     response: HTTPResponse) -> None:25    # pylint: disable=bare-except26    try:27        statsd_client = getattr(request.app.config, 'statsd_client', None)28        if not statsd_client:29            return30        if request.is_single_jrpc:31            statsd_client.from_timings(request.timings)32            statsd_client.from_timings(request.jsonrpc.timings)33            statsd_client.decr('jrpc.inflight')34            statsd_client.gauge('tasks', len(Task.all_tasks()))35            statsd_client._sendbatch()36        elif request.is_batch_jrpc:37            statsd_client.from_timings(request.timings)38            for r in request.jsonrpc:39                statsd_client.from_timings(r.timings)40                statsd_client.decr('jrpc.inflight')41            statsd_client._sendbatch()42    except BaseException as e:43        logger.warning('send_stats', e=e)44@async_nowait_middleware45async def log_stats(request: HTTPRequest,46                    response: HTTPResponse) -> None:47    # pylint: disable=bare-except48    try:49        if request.is_single_jrpc:50            request_timings = fmt_timings(request.timings)51            jsonrpc_timings = fmt_timings(request.jsonrpc.timings)52            logger.debug(53                'log_stats',54                request_timings=request_timings,55                jsonrpc_timings=jsonrpc_timings)56        elif request.is_batch_jrpc:57            request_timings = fmt_timings(request.timings)58            jsonrpc_timings = []59            for r in request.jsonrpc:60                jsonrpc_timings.extend(fmt_timings(r.timings))61            logger.debug('log_stats', request_timings=request_timings,62                         jsonrpc_timings=jsonrpc_timings)63    except BaseException as e:...

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.