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
|
#!/usr/bin/python3
import json
import requests
import subprocess
import signal
import socket
import time
import unittest
USER = "_graphite"
ADDR = "127.0.0.1"
PORT = 8080
class SmokeTest(unittest.TestCase):
@classmethod
def server_ready(cls):
try:
s = socket.socket()
s.connect(
(
ADDR,
PORT,
)
)
except socket.error:
return False
else:
return True
finally:
s.close()
@classmethod
def setUpClass(cls):
# create db
subprocess.check_call(
["/usr/bin/graphite-manage", "migrate", "--run-syncdb"], user=USER
)
# launch dev server
cls.testserver = subprocess.Popen(
[
"/usr/bin/django-admin",
"runserver",
"--settings",
"graphite.settings",
f"{ADDR}:{PORT}",
],
user=USER,
)
# ensure server is ready
retries = 60
while not cls.server_ready() and retries > 0:
time.sleep(0.5)
retries = retries - 1
if retries == 0:
raise RuntimeError("test server is not available")
def test_metrics(self):
with requests.get(f"http://{ADDR}:{PORT}/metrics/index.json") as r:
mlist = json.loads(r.text)
self.assertTrue(mlist == [] or mlist[0])
def test_functions(self):
with requests.get(f"http://{ADDR}:{PORT}/functions") as r:
flist = json.loads(r.text)
self.assertFalse("error" in flist)
self.assertTrue("asPercent" in flist)
@classmethod
def tearDownClass(cls):
cls.testserver.send_signal(signal.SIGTERM)
cls.testserver.wait()
if __name__ == "__main__":
unittest.main()
|
