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
|
import pytest
from openshift_checks.etcd_traffic import EtcdTraffic
@pytest.mark.parametrize('group_names,version,is_active', [
(['oo_masters_to_config'], "3.5", False),
(['oo_masters_to_config'], "3.6", False),
(['oo_nodes_to_config'], "3.4", False),
(['oo_etcd_to_config'], "3.4", True),
(['oo_etcd_to_config'], "1.5", True),
(['oo_etcd_to_config'], "3.1", False),
(['oo_masters_to_config', 'oo_nodes_to_config'], "3.5", False),
(['oo_masters_to_config', 'oo_etcd_to_config'], "3.5", True),
([], "3.4", False),
])
def test_is_active(group_names, version, is_active):
task_vars = dict(
group_names=group_names,
openshift_image_tag=version,
)
assert EtcdTraffic(task_vars=task_vars).is_active() == is_active
@pytest.mark.parametrize('group_names,matched,failed,extra_words', [
(["oo_masters_to_config"], True, True, ["Higher than normal", "traffic"]),
(["oo_masters_to_config", "oo_etcd_to_config"], False, False, []),
(["oo_etcd_to_config"], False, False, []),
])
def test_log_matches_high_traffic_msg(group_names, matched, failed, extra_words):
def execute_module(module_name, *_):
return {
"matched": matched,
"failed": failed,
}
task_vars = dict(
group_names=group_names,
openshift=dict(
common=dict(service_type="origin", is_containerized=False),
)
)
result = EtcdTraffic(execute_module, task_vars).run()
for word in extra_words:
assert word in result.get("msg", "")
assert result.get("failed", False) == failed
@pytest.mark.parametrize('is_containerized,expected_unit_value', [
(False, "etcd"),
(True, "etcd_container"),
])
def test_systemd_unit_matches_deployment_type(is_containerized, expected_unit_value):
task_vars = dict(
openshift=dict(
common=dict(is_containerized=is_containerized),
)
)
def execute_module(module_name, args, *_):
assert module_name == "search_journalctl"
matchers = args["log_matchers"]
for matcher in matchers:
assert matcher["unit"] == expected_unit_value
return {"failed": False}
EtcdTraffic(execute_module, task_vars).run()
|
