<reponame>spotify/aioshumway
# -*- coding: utf-8 -*-
#
# Copyright 2018 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import json

import pytest

import aioshumway


@pytest.fixture
def transport(mocker):
    return mocker.Mock(asyncio.transports.DatagramTransport)


class TestFFWDClientProtocol:
    def test_protocol_sends_msg_on_connect(self, transport):
        msg = 'message'.encode('utf-8')
        proto = aioshumway.FFWDClientProtocol(msg)
        proto.connection_made(transport)

        transport.sendto.assert_called_once_with(msg)

    def test_protocol_closes_transport_after_sending_msg(self, transport):
        msg = 'message'.encode('utf-8')
        proto = aioshumway.FFWDClientProtocol(msg)
        proto.connection_made(transport)

        transport.close.assert_called_once_with()

    def test_protocol_logs_exc(self, mocker):
        logger = mocker.Mock()
        mocker.patch.object(aioshumway, 'logger', new=logger)
        proto = aioshumway.FFWDClientProtocol(b'')
        proto.error_received(Exception('UDP error'))

        logger.warning.assert_called_once_with(
            'Error when sending metric: UDP error\nMetric value: b\'\'')


@pytest.fixture
def coro_tester(mocker):
    mock = mocker.Mock()

    async def _test_coro(*args, **kwargs):
        mock(*args, **kwargs)

    return mock, _test_coro


class TestMeter:
    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        'Meter',
        [aioshumway.Meter, aioshumway.Counter, aioshumway.Timer])
    async def test_aiometers_flush_awaits_coro(self, coro_tester, Meter):
        coro_mock, dummy_coro = coro_tester

        meter = Meter('dummy-what', 'dummy-key')
        await meter.flush(dummy_coro)

        coro_mock.assert_called_once_with({
            'key': meter.key,
            'attributes': meter._attributes,
            'value': meter.value,
            'type': 'metric',
            'tags': meter._tags
        })


@pytest.fixture
def aiometric_relay():
    return aioshumway.MetricRelay('dummy-key', '127.0.0.1', 1000)


class TestMetricRelay:
    @pytest.mark.asyncio
    async def test_relay_uses_provided_event_loop(self, event_loop):
        aiomr = aioshumway.MetricRelay('dummy-key', '127.0.0.1', '1000',
                                       event_loop)
        assert aiomr.loop is event_loop

    @pytest.mark.asyncio
    async def test_relay_gets_its_own_event_loop(self, event_loop):
        aiomr = aioshumway.MetricRelay('dummy-key', '127.0.0.1', '1000')
        assert aiomr.loop is event_loop

    @pytest.mark.asyncio
    async def test_emit_flushes_one_metric(self, mocker, aiometric_relay,
                                           coro_tester):
        coro_mock, dummy_coro = coro_tester
        mocker.patch.object(aiometric_relay, '_sendto', dummy_coro)

        await aiometric_relay.emit('some-metric', 127,
                                   attributes={'internal': True},
                                   tags={'tag1': 256})

        coro_mock.assert_called_once_with({
            'key': 'dummy-key',
            'attributes': {'what': 'some-metric', 'internal': True},
            'value': 127,
            'type': 'metric',
            'tags': {'tag1': 256}
        })

    def test_incr_existing_metric(self, mocker, aiometric_relay):
        counter = mocker.Mock()
        aiometric_relay._metrics['counter1'] = counter

        aiometric_relay.incr('counter1')

        counter.incr.assert_called_once_with(1)

    def test_incr_new_metric(self, mocker, aiometric_relay):
        counter = mocker.Mock()
        mocker.patch.object(
            aioshumway, 'Counter', mocker.Mock(return_value=counter))
        aiometric_relay.incr('counter1')
        counter.incr.assert_called_once_with(1)

    def test_timer_existing_metric(self, mocker, aiometric_relay):
        timer = mocker.Mock()
        aiometric_relay._metrics['timer-timer1'] = timer

        existing_timer = aiometric_relay.timer('timer1')

        assert existing_timer == timer

    def test_timer_new_metric(self, mocker, aiometric_relay):
        timer = mocker.Mock()
        mocker.patch.object(
            aioshumway, 'Timer', mocker.Mock(return_value=timer))

        existing_timer = aiometric_relay.timer('timer1')

        assert existing_timer == timer

    def test_timer_context_man_compatible(self, mocker, aiometric_relay):
        update_method = mocker.Mock()
        timer = aiometric_relay.timer('test-metric')
        mocker.patch.object(timer, 'update', update_method)
        with timer:
            pass
        assert timer._start is not None
        assert isinstance(update_method.call_args[0][0], float)

    @pytest.mark.asyncio
    async def test_flush_flushes_all_metrics(self, mocker, aiometric_relay,
                                             coro_tester):
        coro_mock, dummy_coro = coro_tester
        metric1 = mocker.Mock()
        metric2 = mocker.Mock()

        aiometric_relay._metrics['metric1'] = metric1
        aiometric_relay._metrics['metric2'] = metric2

        mocker.patch.object(aiometric_relay, 'flush_single', dummy_coro)

        await aiometric_relay.flush()

        coro_mock.assert_has_calls([
            mocker.call(metric1),
            mocker.call(metric2)
        ])

    @pytest.mark.asyncio
    async def test_flush_single(self, mocker, aiometric_relay, coro_tester):
        endpoint_mock, endpoint_coro = coro_tester
        mocker.patch.object(aiometric_relay.loop, 'create_datagram_endpoint',
                            endpoint_coro)
        await aiometric_relay.emit(
            'test-metric', 42, attributes={
                'attr1': 'val1'
            })
        call_kwargs = endpoint_mock.call_args[1]
        assert call_kwargs['remote_addr'] == ('127.0.0.1', 1000)

        call_lambda = endpoint_mock.call_args[0][0]
        lambda_result = call_lambda()
        assert lambda_result.message == json.dumps({
            'key': 'dummy-key',
            'attributes': {
                'what': 'test-metric',
                'attr1': 'val1'
            },
            'value': 42,
            'type': 'metric',
            'tags': []
        }).encode('utf-8')

# -*- coding: utf-8 -*-
#
# Copyright 2018 Spotify AB
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import json

import pytest

import aioshumway


@pytest.fixture
def transport(mocker):
    return mocker.Mock(asyncio.transports.DatagramTransport)


class TestFFWDClientProtocol:
    def test_protocol_sends_msg_on_connect(self, transport):
        msg = 'message'.encode('utf-8')
        proto = aioshumway.FFWDClientProtocol(msg)
        proto.connection_made(transport)

        transport.sendto.assert_called_once_with(msg)

    def test_protocol_closes_transport_after_sending_msg(self, transport):
        msg = 'message'.encode('utf-8')
        proto = aioshumway.FFWDClientProtocol(msg)
        proto.connection_made(transport)

        transport.close.assert_called_once_with()

    def test_protocol_logs_exc(self, mocker):
        logger = mocker.Mock()
        mocker.patch.object(aioshumway, 'logger', new=logger)
        proto = aioshumway.FFWDClientProtocol(b'')
        proto.error_received(Exception('UDP error'))

        logger.warning.assert_called_once_with(
            'Error when sending metric: UDP error\nMetric value: b\'\'')


@pytest.fixture
def coro_tester(mocker):
    mock = mocker.Mock()

    async def _test_coro(*args, **kwargs):
        mock(*args, **kwargs)

    return mock, _test_coro


class TestMeter:
    @pytest.mark.asyncio
    @pytest.mark.parametrize(
        'Meter',
        [aioshumway.Meter, aioshumway.Counter, aioshumway.Timer])
    async def test_aiometers_flush_awaits_coro(self, coro_tester, Meter):
        coro_mock, dummy_coro = coro_tester

        meter = Meter('dummy-what', 'dummy-key')
        await meter.flush(dummy_coro)

        coro_mock.assert_called_once_with({
            'key': meter.key,
            'attributes': meter._attributes,
            'value': meter.value,
            'type': 'metric',
            'tags': meter._tags
        })


@pytest.fixture
def aiometric_relay():
    return aioshumway.MetricRelay('dummy-key', '127.0.0.1', 1000)


class TestMetricRelay:
    @pytest.mark.asyncio
    async def test_relay_uses_provided_event_loop(self, event_loop):
        aiomr = aioshumway.MetricRelay('dummy-key', '127.0.0.1', '1000',
                                       event_loop)
        assert aiomr.loop is event_loop

    @pytest.mark.asyncio
    async def test_relay_gets_its_own_event_loop(self, event_loop):
        aiomr = aioshumway.MetricRelay('dummy-key', '127.0.0.1', '1000')
        assert aiomr.loop is event_loop

    @pytest.mark.asyncio
    async def test_emit_flushes_one_metric(self, mocker, aiometric_relay,
                                           coro_tester):
        coro_mock, dummy_coro = coro_tester
        mocker.patch.object(aiometric_relay, '_sendto', dummy_coro)

        await aiometric_relay.emit('some-metric', 127,
                                   attributes={'internal': True},
                                   tags={'tag1': 256})

        coro_mock.assert_called_once_with({
            'key': 'dummy-key',
            'attributes': {'what': 'some-metric', 'internal': True},
            'value': 127,
            'type': 'metric',
            'tags': {'tag1': 256}
        })

    def test_incr_existing_metric(self, mocker, aiometric_relay):
        counter = mocker.Mock()
        aiometric_relay._metrics['counter1'] = counter

        aiometric_relay.incr('counter1')

        counter.incr.assert_called_once_with(1)

    def test_incr_new_metric(self, mocker, aiometric_relay):
        counter = mocker.Mock()
        mocker.patch.object(
            aioshumway, 'Counter', mocker.Mock(return_value=counter))
        aiometric_relay.incr('counter1')
        counter.incr.assert_called_once_with(1)

    def test_timer_existing_metric(self, mocker, aiometric_relay):
        timer = mocker.Mock()
        aiometric_relay._metrics['timer-timer1'] = timer

        existing_timer = aiometric_relay.timer('timer1')

        assert existing_timer == timer

    def test_timer_new_metric(self, mocker, aiometric_relay):
        timer = mocker.Mock()
        mocker.patch.object(
            aioshumway, 'Timer', mocker.Mock(return_value=timer))

        existing_timer = aiometric_relay.timer('timer1')

        assert existing_timer == timer

    def test_timer_context_man_compatible(self, mocker, aiometric_relay):
        update_method = mocker.Mock()
        timer = aiometric_relay.timer('test-metric')
        mocker.patch.object(timer, 'update', update_method)
        with timer:
            pass
        assert timer._start is not None
        assert isinstance(update_method.call_args[0][0], float)

    @pytest.mark.asyncio
    async def test_flush_flushes_all_metrics(self, mocker, aiometric_relay,
                                             coro_tester):
        coro_mock, dummy_coro = coro_tester
        metric1 = mocker.Mock()
        metric2 = mocker.Mock()

        aiometric_relay._metrics['metric1'] = metric1
        aiometric_relay._metrics['metric2'] = metric2

        mocker.patch.object(aiometric_relay, 'flush_single', dummy_coro)

        await aiometric_relay.flush()

        coro_mock.assert_has_calls([
            mocker.call(metric1),
            mocker.call(metric2)
        ])

    @pytest.mark.asyncio
    async def test_flush_single(self, mocker, aiometric_relay, coro_tester):
        endpoint_mock, endpoint_coro = coro_tester
        mocker.patch.object(aiometric_relay.loop, 'create_datagram_endpoint',
                            endpoint_coro)
        await aiometric_relay.emit(
            'test-metric', 42, attributes={
                'attr1': 'val1'
            })
        call_kwargs = endpoint_mock.call_args[1]
        assert call_kwargs['remote_addr'] == ('127.0.0.1', 1000)

        call_lambda = endpoint_mock.call_args[0][0]
        lambda_result = call_lambda()
        assert lambda_result.message == json.dumps({
            'key': 'dummy-key',
            'attributes': {
                'what': 'test-metric',
                'attr1': 'val1'
            },
            'value': 42,
            'type': 'metric',
            'tags': []
        }).encode('utf-8')
# -*- coding: utf-8 -*- # # Copyright 2018 Spotify AB # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License.