-
Notifications
You must be signed in to change notification settings - Fork 4.9k
/
test_mysql.py
48 lines (37 loc) · 1.36 KB
/
test_mysql.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
import metricbeat
import os
import pytest
import sys
import unittest
MYSQL_FIELDS = metricbeat.COMMON_FIELDS + ["mysql"]
MYSQL_STATUS_FIELDS = ["clients", "cluster", "cpu", "keyspace", "memory",
"persistence", "replication", "server", "stats"]
@metricbeat.parameterized_with_supported_versions
class Test(metricbeat.BaseTest):
COMPOSE_SERVICES = ['mysql']
@unittest.skipUnless(metricbeat.INTEGRATION_TESTS, "integration test")
@pytest.mark.tag('integration')
def test_status(self):
"""
MySQL module outputs an event.
"""
self.render_config_template(modules=[{
"name": "mysql",
"metricsets": ["status"],
"hosts": self.get_hosts(),
"period": "5s"
}])
proc = self.start_beat()
self.wait_until(lambda: self.output_lines() > 0)
proc.check_kill_and_wait()
self.assert_no_logged_warnings()
output = self.read_output_json()
self.assertEqual(len(output), 1)
evt = output[0]
self.assertCountEqual(self.de_dot(MYSQL_FIELDS), evt.keys(), evt)
status = evt["mysql"]["status"]
assert status["connections"] > 0
assert status["opened_tables"] > 0
self.assert_fields_are_documented(evt)
def get_hosts(self):
return ['root:test@tcp({})/'.format(self.compose_host())]