statsd_client_python/tests/test_formats.py at d2474de5513c9062f11a3f129f85d085d68a1d63 · lirsacc/statsd_client_python · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
from collections.abc import Mapping

import pytest

from statsd.exceptions import InvalidTags
from statsd.formats import (
    DogstatsdSerializer,
    GraphiteSerializer,
    TelegrafSerializer,
)


SerializeInput = tuple[str, str, str, float, Mapping[str, str]]


CASE_SIMPLE: SerializeInput = ("my_metric", "c", "1", 1, {})
CASE_TAG_AND_RATE: SerializeInput = (
    "my_metric",
    "ms",
    "123456",
    0.4,
    {"foo": "1", "bar": "some_value"},
)
CASE_TAG_NO_RATE: SerializeInput = (
    "my_metric",
    "ms",
    "123456",
    1,
    {"foo": "1", "bar": "some_value"},
)
CASE_EMPTY_VALUE: SerializeInput = (
    "my_metric",
    "ms",
    "123456",
    1,
    {"foo": "1", "bar": "", "baz": "some_value"},
)
CASE_INVALID_CHARS = SerializeInput = (
    "my_metric",
    "ms",
    "123456",
    1,
    {"foo": "*.:=1foo"},
)
CASE_EMPTY_KEY = SerializeInput = (
    "my_metric",
    "ms",
    "123456",
    1,
    {"foo": "1", "": "some_value"},
)


@pytest.mark.parametrize(
    ("params", "expected"),
    [
        (CASE_SIMPLE, "my_metric:1|c"),
        (CASE_TAG_AND_RATE, "my_metric:123456|ms|@0.4|#foo:1,bar:some_value"),
        (CASE_TAG_NO_RATE, "my_metric:123456|ms|#foo:1,bar:some_value"),
        (CASE_EMPTY_VALUE, "my_metric:123456|ms|#foo:1,bar,baz:some_value"),
        (CASE_INVALID_CHARS, "my_metric:123456|ms|#foo:-.--1foo"),
        (CASE_EMPTY_KEY, "my_metric:123456|ms|#foo:1"),
    ],
)
def test_dogstatsd_format_ok(params: SerializeInput, expected: str) -> None:
    assert DogstatsdSerializer().serialize(*params) == expected


@pytest.mark.parametrize(
    ("params", "expected"),
    [
        (CASE_SIMPLE, "my_metric:1|c"),
        (CASE_TAG_AND_RATE, "my_metric,foo=1,bar=some_value:123456|ms|@0.4"),
        (CASE_TAG_NO_RATE, "my_metric,foo=1,bar=some_value:123456|ms"),
        (CASE_INVALID_CHARS, "my_metric,foo=-.--1foo:123456|ms"),
        (CASE_EMPTY_KEY, "my_metric,foo=1:123456|ms"),
    ],
)
def test_telegraf_format_ok(params: SerializeInput, expected: str) -> None:
    assert TelegrafSerializer().serialize(*params) == expected


@pytest.mark.parametrize("params", [CASE_EMPTY_VALUE])
def test_telegraf_format_invalid(params: SerializeInput) -> None:
    with pytest.raises(InvalidTags):
        TelegrafSerializer().serialize(*params)


@pytest.mark.parametrize(
    ("params", "expected"),
    [
        (CASE_SIMPLE, "my_metric:1|c"),
        (CASE_TAG_AND_RATE, "my_metric;foo=1;bar=some_value:123456|ms|@0.4"),
        (CASE_TAG_NO_RATE, "my_metric;foo=1;bar=some_value:123456|ms"),
        (CASE_INVALID_CHARS, "my_metric;foo=-.--1foo:123456|ms"),
        (CASE_EMPTY_KEY, "my_metric;foo=1:123456|ms"),
    ],
)
def test_graphite_format_ok(params: SerializeInput, expected: str) -> None:
    assert GraphiteSerializer().serialize(*params) == expected


@pytest.mark.parametrize("params", [CASE_EMPTY_VALUE])
def test_graphite_format_invalid(params: SerializeInput) -> None:
    with pytest.raises(InvalidTags):
        GraphiteSerializer().serialize(*params)