forked from OpenTSDB/tcollector
-
Notifications
You must be signed in to change notification settings - Fork 1
/
tests.py
executable file
·108 lines (90 loc) · 3.87 KB
/
tests.py
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
107
108
#!/usr/bin/python
# This file is part of tcollector.
# Copyright (C) 2013 The tcollector Authors.
#
# This program 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 3 of the License, or (at your
# option) any later version. This program 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 this program. If not,
# see <http://www.gnu.org/licenses/>.
import os
import sys
from stat import S_ISDIR, S_ISREG, ST_MODE
import unittest
import tcollector
class CollectorsTests(unittest.TestCase):
def test_collectorsAccessRights(self):
"""Test of collectors access rights, permissions should be 0100775."""
def check_access_rights(top):
for f in os.listdir(top):
pathname = os.path.join(top, f)
mode = os.stat(pathname).st_mode
if S_ISDIR(mode):
# directory, recurse into it
check_access_rights(pathname)
elif S_ISREG(mode):
# file, check permissions
self.assertEqual("0100775", oct(os.stat(pathname)[ST_MODE]))
else:
# unknown file type
pass
collectors_path = os.path.dirname(os.path.abspath(__file__)) + \
"/collectors/0"
check_access_rights(collectors_path)
class TSDBlacklistingTests(unittest.TestCase):
"""
Tests of TSD blacklisting logic
https://github.com/OpenTSDB/tcollector/commit/c191d0d0889860db2ea231cad02e398843031a74
"""
def setUp(self):
# Stub out the randomness
self.random_shuffle = tcollector.random.shuffle
tcollector.random.shuffle = lambda x: x
def tearDown(self):
tcollector.random.shuffle = self.random_shuffle
def mkSenderThread(self, tsds):
return tcollector.SenderThread(None, True, tsds, False, {})
def test_blacklistOneConnection(self):
tsd = ("localhost", 4242)
sender = self.mkSenderThread([tsd])
sender.pick_connection()
self.assertEqual(tsd, (sender.host, sender.port))
sender.blacklist_connection()
sender.pick_connection()
self.assertEqual(tsd, (sender.host, sender.port))
def test_blacklistTwoConnections(self):
tsd1 = ("localhost", 4242)
tsd2 = ("localhost", 4243)
sender = self.mkSenderThread([tsd1, tsd2])
sender.pick_connection()
self.assertEqual(tsd1, (sender.host, sender.port))
sender.blacklist_connection()
sender.pick_connection()
self.assertEqual(tsd2, (sender.host, sender.port))
sender.blacklist_connection()
sender.pick_connection()
self.assertEqual(tsd1, (sender.host, sender.port))
def test_doublePickOneConnection(self):
tsd = ("localhost", 4242)
sender = self.mkSenderThread([tsd])
sender.pick_connection()
self.assertEqual(tsd, (sender.host, sender.port))
sender.pick_connection()
self.assertEqual(tsd, (sender.host, sender.port))
def test_doublePickTwoConnections(self):
tsd1 = ("localhost", 4242)
tsd2 = ("localhost", 4243)
sender = self.mkSenderThread([tsd1, tsd2])
sender.pick_connection()
self.assertEqual(tsd1, (sender.host, sender.port))
sender.pick_connection()
self.assertEqual(tsd2, (sender.host, sender.port))
sender.pick_connection()
self.assertEqual(tsd1, (sender.host, sender.port))
if __name__ == '__main__':
tcollector.setup_logging()
sys.exit(unittest.main())