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

How to use _record_property_statsd method in pytest-play

Best Python code snippet using pytest-play_python

test_metrics.py

Source:test_metrics.py

1#!/usr/bin/env python2# -*- coding: utf-8 -*-3"""Tests for `pytest_play` package."""4import pytest5def test_record_elapsed():6    import mock7    mock_engine = mock.MagicMock()8    elapsed = 0.1259    mock_engine.variables = {'_elapsed': elapsed}10    mock_record_property = mock.MagicMock()11    mock_record_property_statsd = mock.MagicMock()12    from pytest_play import providers13    provider = providers.MetricsProvider(mock_engine)14    provider._record_property = mock_record_property15    provider._record_property_statsd = mock_record_property_statsd16    assert provider.engine is mock_engine17    provider.command_record_elapsed({18        'provider': 'metrics',19        'type': 'record_elapsed',20        'name': 'previous_command',21        'comment': 'record last command elapsed time '22        'under key name previous command'23    })24    assert mock_engine \25        .update_variables \26        .assert_called_once_with(27            {'previous_command': elapsed}) is None28    assert mock_record_property \29        .assert_called_once_with(30            'previous_command', elapsed) is None31    assert mock_record_property_statsd \32        .assert_called_once_with(33            'previous_command', elapsed, metric_type='timing',34            meas_unit='s') is None35def test_record_elapsed_statsd():36    import mock37    mock_engine = mock.MagicMock()38    elapsed = 0.12539    mock_engine.variables = {'_elapsed': elapsed}40    mock_record_property = mock.MagicMock()41    from pytest_play import providers42    with mock.patch(43            'pytest_play.providers.metrics.MetricsProvider.statsd_client',44            new_callable=mock.PropertyMock) as statsd_client:45        provider = providers.MetricsProvider(mock_engine)46        provider._record_property = mock_record_property47        assert provider.engine is mock_engine48        provider.command_record_elapsed({49            'provider': 'metrics',50            'type': 'record_elapsed',51            'name': 'previous_command',52            'comment': 'record last command elapsed time '53            'under key name previous command'54        })55        assert statsd_client.return_value.timing.assert_called_once() is None56def test_record_elapsed_key_error():57    import mock58    mock_engine = mock.MagicMock()59    mock_engine.variables = {}60    from pytest_play import providers61    provider = providers.MetricsProvider(mock_engine)62    assert provider.engine is mock_engine63    with pytest.raises(KeyError):64        provider.command_record_elapsed({65            'provider': 'metrics',66            'type': 'record_elapsed',67            'name': 'previous_command',68            'comment': 'record last command elapsed time '69            'under key name previous command'70        })71def test_record_property():72    import mock73    mock_engine = mock.MagicMock()74    elapsed = 0.12575    mock_engine.variables = {'_elapsed': elapsed}76    mock_record_property = mock.MagicMock()77    mock_record_property_statsd = mock.MagicMock()78    from pytest_play import providers79    provider = providers.MetricsProvider(mock_engine)80    provider._record_property = mock_record_property81    provider._record_property_statsd = mock_record_property_statsd82    assert provider.engine is mock_engine83    mock_engine.execute_command.return_value = elapsed*100084    provider.command_record_property({85        'provider': 'metrics',86        'type': 'record_property',87        'name': 'elapsed_milliseconds',88        'expression': 'variables["_elapsed"]*1000',89    })90    assert mock_engine.execute_command.assert_called_once_with(91        {'provider': 'python',92         'type': 'exec',93         'expression': 'variables["_elapsed"]*1000'}) is None94    assert mock_engine \95        .update_variables \96        .assert_called_once_with(97            {'elapsed_milliseconds': elapsed*1000}) is None98    assert mock_record_property \99        .assert_called_once_with(100            'elapsed_milliseconds', elapsed*1000) is None101    assert mock_record_property_statsd \102        .assert_called_once_with(103            'elapsed_milliseconds', elapsed*1000,104            metric_type=None, meas_unit=None) is None105def test_record_property_string_no_statsd():106    import mock107    mock_engine = mock.MagicMock()108    mock_record_property = mock.MagicMock()109    mock_record_property_statsd = mock.MagicMock()110    from pytest_play import providers111    provider = providers.MetricsProvider(mock_engine)112    provider._record_property = mock_record_property113    provider._record_property_statsd = mock_record_property_statsd114    assert provider.engine is mock_engine115    mock_engine.execute_command.return_value = "a string"116    with mock.patch(117            'pytest_play.providers.metrics.MetricsProvider.statsd_client',118            new_callable=mock.PropertyMock) as statsd_client:119        provider.command_record_property({120            'provider': 'metrics',121            'type': 'record_property',122            'name': 'elapsed_milliseconds',123            'expression': '"a string"',124        })125        assert mock_engine.execute_command.assert_called_once_with(126            {'provider': 'python',127             'type': 'exec',128             'expression': '"a string"'}) is None129        assert mock_engine \130            .update_variables \131            .assert_called_once_with(132                {'elapsed_milliseconds': "a string"}) is None133        assert mock_record_property \134            .assert_called_once_with(135                'elapsed_milliseconds', "a string") is None136        assert statsd_client.called is False137def test_record_property_statsd_no_metric_type():138    import mock139    mock_engine = mock.MagicMock()140    elapsed = 0.125141    mock_engine.variables = {'_elapsed': elapsed}142    mock_record_property = mock.MagicMock()143    from pytest_play import providers144    with mock.patch(145            'pytest_play.providers.metrics.MetricsProvider.statsd_client',146            new_callable=mock.PropertyMock) as statsd_client:147        provider = providers.MetricsProvider(mock_engine)148        provider._record_property = mock_record_property149        assert provider.engine is mock_engine150        mock_engine.execute_command.return_value = elapsed*1000151        provider.command_record_property({152            'provider': 'metrics',153            'type': 'record_property',154            'name': 'elapsed_milliseconds',155            'expression': 'variables["_elapsed"]*1000',156        })157        assert mock_engine.execute_command.assert_called_once_with(158            {'provider': 'python',159             'type': 'exec',160             'expression': 'variables["_elapsed"]*1000'}) is None161        assert mock_engine \162            .update_variables \163            .assert_called_once_with(164                {'elapsed_milliseconds': elapsed*1000}) is None165        assert mock_record_property \166            .assert_called_once_with(167                'elapsed_milliseconds', elapsed*1000) is None168        assert statsd_client.return_value.timing.called is False169def test_record_property_statsd_metric_type_timing():170    import mock171    mock_engine = mock.MagicMock()172    elapsed = 0.125173    mock_engine.variables = {'_elapsed': elapsed}174    mock_record_property = mock.MagicMock()175    from pytest_play import providers176    with mock.patch(177            'pytest_play.providers.metrics.MetricsProvider.statsd_client',178            new_callable=mock.PropertyMock) as statsd_client:179        provider = providers.MetricsProvider(mock_engine)180        provider._record_property = mock_record_property181        assert provider.engine is mock_engine182        mock_engine.execute_command.return_value = elapsed*1000183        provider.command_record_property({184            'provider': 'metrics',185            'type': 'record_property',186            'name': 'elapsed_milliseconds',187            'expression': 'variables["_elapsed"]*1000',188            'metric_type': 'timing',189        })190        assert mock_engine.execute_command.assert_called_once_with(191            {'provider': 'python',192             'type': 'exec',193             'expression': 'variables["_elapsed"]*1000'}) is None194        assert mock_engine \195            .update_variables \196            .assert_called_once_with(197                {'elapsed_milliseconds': elapsed*1000}) is None198        assert mock_record_property \199            .assert_called_once_with(200                'elapsed_milliseconds', elapsed*1000) is None201        assert statsd_client.return_value.timing \202            .assert_called_once_with(203                'elapsed_milliseconds', elapsed*1000) is None204@pytest.mark.parametrize("metric_type", ['timing', 'gauge', ])205def test_record_property_statsd_metric_type_negative(metric_type):206    import mock207    mock_engine = mock.MagicMock()208    mock_record_property = mock.MagicMock()209    from pytest_play import providers210    with mock.patch(211            'pytest_play.providers.metrics.MetricsProvider.statsd_client',212            new_callable=mock.PropertyMock) as statsd_client:213        provider = providers.MetricsProvider(mock_engine)214        provider._record_property = mock_record_property215        assert provider.engine is mock_engine216        mock_engine.execute_command.return_value = "a string"217        with pytest.raises(ValueError):218            provider.command_record_property({219                'provider': 'metrics',220                'type': 'record_property',221                'name': 'elapsed_milliseconds',222                'expression': '"a string"',223                'metric_type': metric_type,224            })225        assert mock_engine.execute_command.assert_called_once_with(226            {'provider': 'python',227             'type': 'exec',228             'expression': '"a string"'}) is None229        assert mock_engine \230            .update_variables \231            .called is False232        assert mock_record_property \233            .called is False234        assert statsd_client.return_value.timing \235            .called is False236@pytest.mark.parametrize("metric_type", ['TIMING', ])237def test_record_property_statsd_metric_type_negative_invalid(metric_type):238    import mock239    mock_engine = mock.MagicMock()240    mock_record_property = mock.MagicMock()241    from pytest_play import providers242    with mock.patch(243            'pytest_play.providers.metrics.MetricsProvider.statsd_client',244            new_callable=mock.PropertyMock) as statsd_client:245        provider = providers.MetricsProvider(mock_engine)246        provider._record_property = mock_record_property247        assert provider.engine is mock_engine248        mock_engine.execute_command.return_value = "1"249        with pytest.raises(ValueError):250            provider.command_record_property({251                'provider': 'metrics',252                'type': 'record_property',253                'name': 'elapsed_milliseconds',254                'expression': '"1"',255                'metric_type': metric_type,256            })257        assert mock_engine.execute_command.assert_called_once_with(258            {'provider': 'python',259             'type': 'exec',260             'expression': '"1"'}) is None261        assert mock_engine \262            .update_variables \263            .called is False264        assert mock_record_property \265            .called is False266        assert statsd_client.return_value.timing \267            .called is False268def test_record_property_statsd_metric_type_gauge():269    import mock270    mock_engine = mock.MagicMock()271    elapsed = 0.125272    mock_engine.variables = {'_elapsed': elapsed}273    mock_record_property = mock.MagicMock()274    from pytest_play import providers275    with mock.patch(276            'pytest_play.providers.metrics.MetricsProvider.statsd_client',277            new_callable=mock.PropertyMock) as statsd_client:278        provider = providers.MetricsProvider(mock_engine)279        provider._record_property = mock_record_property280        assert provider.engine is mock_engine281        mock_engine.execute_command.return_value = elapsed*1000282        provider.command_record_property({283            'provider': 'metrics',284            'type': 'record_property',285            'name': 'elapsed_milliseconds',286            'expression': 'variables["_elapsed"]*1000',287            'metric_type': 'gauge',288        })289        assert mock_engine.execute_command.assert_called_once_with(290            {'provider': 'python',291             'type': 'exec',292             'expression': 'variables["_elapsed"]*1000'}) is None293        assert mock_engine \294            .update_variables \295            .assert_called_once_with(296                {'elapsed_milliseconds': elapsed*1000}) is None297        assert mock_record_property \298            .assert_called_once_with(299                'elapsed_milliseconds', elapsed*1000) is None300        assert statsd_client.return_value.gauge \301            .assert_called_once_with(302                'elapsed_milliseconds', elapsed*1000) is None303def test_record_elapsed_start():304    import mock305    mock_engine = mock.MagicMock()306    mock_engine.variables = {}307    from pytest_play import providers308    provider = providers.MetricsProvider(mock_engine)309    assert provider.engine is mock_engine310    time_start = 1550770816.1716287311    with mock.patch('pytest_play.providers.metrics.time') as time:312        time.time.return_value = time_start313        provider.command_record_elapsed_start({314            'provider': 'metrics',315            'type': 'record_elapsed_start',316            'name': 'async_update',317        })318    assert mock_engine.update_variables.assert_called_once_with(319        {'async_update': time_start}) is None320def test_record_elapsed_stop():321    import mock322    mock_engine = mock.MagicMock()323    time_start = 1550770816.1716287324    mock_engine.variables = {'async_update': time_start}325    mock_record_property = mock.MagicMock()326    mock_record_property_statsd = mock.MagicMock()327    from pytest_play import providers328    provider = providers.MetricsProvider(mock_engine)329    provider._record_property = mock_record_property330    provider._record_property_statsd = mock_record_property_statsd331    assert provider.engine is mock_engine332    time_stop = 1550770817.1716287333    with mock.patch('pytest_play.providers.metrics.time') as time:334        time.time.return_value = time_stop335        provider.command_record_elapsed_stop({336            'provider': 'metrics',337            'type': 'record_elapsed_stop',338            'name': 'async_update',339        })340    assert mock_engine.update_variables.assert_called_once_with(341        {'async_update': time_stop-time_start}) is None342def test_record_elapsed_stop_statsd():343    import mock344    mock_engine = mock.MagicMock()345    time_start = 1550770816.1716287346    mock_engine.variables = {'async_update': time_start}347    mock_record_property = mock.MagicMock()348    from pytest_play import providers349    with mock.patch(350            'pytest_play.providers.metrics.MetricsProvider.statsd_client',351            new_callable=mock.PropertyMock) as statsd_client:352        provider = providers.MetricsProvider(mock_engine)353        provider._record_property = mock_record_property354        assert provider.engine is mock_engine355        time_stop = 1550770817.1716287356        with mock.patch('pytest_play.providers.metrics.time') as time:357            time.time.return_value = time_stop358            provider.command_record_elapsed_stop({359                'provider': 'metrics',360                'type': 'record_elapsed_stop',361                'name': 'async_update',362            })363        assert statsd_client.return_value.timing.assert_called_once_with(364            'async_update', (time_stop-time_start)*1000) is None365def test_record_elapsed_stop_key_error():366    import mock367    mock_engine = mock.MagicMock()368    mock_engine.variables = {}369    from pytest_play import providers370    provider = providers.MetricsProvider(mock_engine)371    assert provider.engine is mock_engine372    with pytest.raises(KeyError):373        provider.command_record_elapsed_stop({374            'provider': 'metrics',375            'type': 'record_elapsed_stop',376            'name': 'async_update',...

metrics.py

Source:metrics.py

...13        host = self.engine.request.config.getoption('stats_host')14        port = self.engine.request.config.getoption('stats_port')15        prefix = self.engine.request.config.getoption('stats_prefix')16        return statsd.StatsClient(host, port, prefix=prefix)17    def _record_property_statsd(18            self, name, value, metric_type=None, meas_unit=None):19        if STATSD:20            if metric_type is not None:21                statsd_client = self.statsd_client22                method = None23                if metric_type == 'timing':24                    method = statsd_client.timing25                    if meas_unit == 's':26                        value = value * 100027                    assert float(value)28                elif metric_type == 'gauge':29                    method = statsd_client.gauge30                    assert float(value)31                else:32                    raise ValueError("metric_type not valid: ", metric_type)33                if method is not None:34                    method(name, value)35    def record_property(self, name, value, metric_type=None, meas_unit=None):36        """ Record a property metrics """37        self._record_property_statsd(38            name, value, metric_type=metric_type, meas_unit=meas_unit)39        self._record_property(name, value)40    def command_record_property(self, command, **kwargs):41        """ record a property (dynamic expression) """42        name = command['name']43        expression = command['expression']44        metric_type = command.get('metric_type', None)45        value = self.engine.execute_command(46            {'provider': 'python',47             'type': 'exec',48             'expression': expression})49        self.record_property(name, value, metric_type=metric_type)50        self.engine.update_variables({name: value})51    def command_record_elapsed(self, command, **kwargs):...

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.