# -*- coding: utf-8 -*-

# copyright 2018 LOGILAB S.A. (Paris, FRANCE), all rights reserved.

# contact http://www.logilab.fr/ -- mailto:contact@logilab.fr

#

# This file is part of CubicWeb.

#

# CubicWeb is free software: you can redistribute it and/or modify it under the

# terms of the GNU Lesser General Public License as published by the Free

# Software Foundation, either version 2.1 of the License, or (at your option)

# any later version.

#

# CubicWeb is distributed in the hope that it will be useful, but WITHOUT

# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS

# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more

# details.

#

# You should have received a copy of the GNU Lesser General Public License along

# with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.

"""unit tests for module cubicweb.statsd_logger"""

import threading

import socket

import time

import re

from unittest import TestCase

from cubicweb import statsd_logger as statsd

UDP_PORT = None

RUNNING = True

SOCK = socket.socket(socket.AF_INET,

                     socket.SOCK_DGRAM)

SOCK.settimeout(0.1)

STATSD = None

DATA = []

def statsd_rcv():

    while RUNNING:

        try:

            data, addr = SOCK.recvfrom(1024)

            if data:

                rcv = [row.strip().decode() for row in data.splitlines()]

                DATA.extend(rcv)

        except socket.timeout:

            pass

def setUpModule(*args):

    global UDP_PORT, STATSD

    SOCK.bind(('127.0.0.1', 0))

    UDP_PORT = SOCK.getsockname()[1]

    STATSD = threading.Thread(target=statsd_rcv)

    STATSD.start()

    statsd.setup('test', ('127.0.0.1', UDP_PORT))

def tearDownModule(*args):

    global RUNNING

    RUNNING = False

    STATSD.join()

    statsd.teardown()

class StatsdTC(TestCase):

    def setUp(self):

        super(StatsdTC, self).setUp()

        DATA[:] = []

    def check_received(self, value):

        for i in range(10):

            if value in DATA:

                break

            time.sleep(0.01)

        else:

            self.assertIn(value, DATA)

    def check_received_ms(self, value):

        value = re.compile(value.replace('?', r'\d'))

        for i in range(10):

            if [x for x in DATA if value.match(x)]:

                break

            time.sleep(0.01)

        else:

            self.assertTrue([x for x in DATA if value.match(x)], DATA)

    def test_statsd_c(self):

        statsd.statsd_c('context')

        self.check_received('test.context:1|c')

        statsd.statsd_c('context', 10)

        self.check_received('test.context:10|c')

    def test_statsd_g(self):

        statsd.statsd_g('context', 42)

        self.check_received('test.context:42|g')

        statsd.statsd_g('context', 'Igorrr')

        self.check_received('test.context:Igorrr|g')

    def test_statsd_t(self):

        statsd.statsd_t('context', 1)

        self.check_received('test.context:1.0000|ms')

        statsd.statsd_t('context', 10)

        self.check_received('test.context:10.0000|ms')

        statsd.statsd_t('context', 0.12344)

        self.check_received('test.context:0.1234|ms')

        statsd.statsd_t('context', 0.12345)

        self.check_received('test.context:0.1235|ms')

    def test_decorator(self):

        @statsd.statsd_timeit

        def measure_me_please():

            "some nice function"

            return 42

        self.assertEqual(measure_me_please.__doc__,

                         "some nice function")

        measure_me_please()

        self.check_received_ms('test.measure_me_please:0.0???|ms')

        self.check_received('test.measure_me_please:1|c')

    def test_context_manager(self):

        with statsd.statsd_timethis('cm'):

            time.sleep(0.1)

        self.check_received_ms('test.cm:100.????|ms')

        self.check_received('test.cm:1|c')

if __name__ == '__main__':

    from unittest import main

    main()