summaryrefslogtreecommitdiffstats
path: root/gstatsd/service_test.py
blob: 1b6bc2d345c9050134b85e8a6d9ee8aa76e9ef93 (plain)
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

# standard
import unittest

# local
from gstatsd import service


class StatsServiceTest(unittest.TestCase):

    def setUp(self):
        args = (':8125', [':9100'], 5, 90, 0)
        self.svc = service.StatsDaemon(*args)

    def test_construct(self):
        svc = service.StatsDaemon('8125', ['9100'], 5, 90, 0)
        self.assertEquals(svc._bindaddr, ('', 8125))
        self.assertEquals(svc._interval, 5.0)
        self.assertEquals(svc._percent, 90.0)
        self.assertEquals(svc._debug, 0)
        self.assertEquals(svc._targets[0], (svc._send_graphite, ('', 9100)))

        svc = service.StatsDaemon('bar:8125', ['foo:9100'], 5, 90, 1)
        self.assertEquals(svc._bindaddr, ('bar', 8125))
        self.assertEquals(svc._targets[0], (svc._send_graphite, ('foo', 9100)))
        self.assertEquals(svc._debug, 1)

    def test_backend(self):
        service.StatsDaemon._send_foo = lambda self, x, y: None
        svc = service.StatsDaemon('8125', ['foo:bar:9100'], 5, 90, 0)
        self.assertEquals(svc._targets[0], (svc._send_foo, ('bar', 9100)))

    def test_counters(self):
        pkt = 'foo:1|c'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._counts, {'foo': 1})
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._counts, {'foo': 2})
        pkt = 'foo:-1|c'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._counts, {'foo': 1})

    def test_counters_sampled(self):
        pkt = 'foo:1|c|@.5'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._counts, {'foo': 2})

    def test_timers(self):
        pkt = 'foo:20|ms'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._timers, {'foo': [20.0]})
        pkt = 'foo:10|ms'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._timers, {'foo': [20.0, 10.0]})

    def test_key_sanitize(self):
        pkt = '\t\n#! foo . bar \0 ^:1|c'
        self.svc._process(pkt, None)
        self.assertEquals(self.svc._counts, {'foo.bar': 1})


def main():
    unittest.main()


if __name__ == '__main__':
    main()