forked from Checkmk/checkmk
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathtest_ibm_mq_channels.py
209 lines (176 loc) · 7.55 KB
/
test_ibm_mq_channels.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
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
#!/usr/bin/env python3
# Copyright (C) 2019 Checkmk GmbH - License: GNU General Public License v2
# This file is part of Checkmk (https://checkmk.com). It is subject to the terms and
# conditions defined in the file COPYING, which is part of this source code package.
from typing import Any
import pytest
from cmk.agent_based.v2 import IgnoreResultsError
from cmk.plugins.collection.agent_based.ibm_mq_channels import parse_ibm_mq_channels
from .checktestlib import Check
from .test_ibm_mq_include import parse_info
pytestmark = pytest.mark.checks
CHECK_NAME = "ibm_mq_channels"
def test_parse() -> None:
lines = """\
QMNAME(MY.TEST) STATUS(RUNNING) NOW(2020-04-03T17:27:02+0200)
5724-H72 (C) Copyright IBM Corp. 1994, 2015.
Starting MQSC for queue manager MY.TEST.
AMQ8414: Display Channel details.
CHANNEL(HERE.TO.THERE.ONE) CHLTYPE(SDR)
XMITQ(HERE.TO.THERE.ONE.XMIT)
AMQ8414: Display Channel details.
CHANNEL(HERE.TO.THERE.TWO) CHLTYPE(SDR)
XMITQ(HERE.TO.THERE.TWO.XMIT)
AMQ8414: Display Channel details.
CHANNEL(SYSTEM.DEF.SENDER) CHLTYPE(SDR)
XMITQ( )
AMQ8417: Display Channel Status details.
CHANNEL(HERE.TO.THERE.TWO) CHLTYPE(SDR)
COMPRATE(0,0) COMPTIME(0,0)
CONNAME(55.888.222.333(1414),22,333.444.555(1414))
EXITTIME(0,0) MONCHL(OFF)
CURRENT RQMNAME( )
STATUS(RETRYING) SUBSTATE( )
XBATCHSZ(0,0) XMITQ(HERE.TO.THERE.TWO.XMIT)
XQTIME(0,1)
AMQ8417: Display Channel Status details.
CHANNEL(HERE.TO.THERE.ONE) CHLTYPE(SDR)
COMPRATE(0,0) COMPTIME(0,0)
CONNAME(62.240.197.243(1414),62.240.197.244(1414))
EXITTIME(0,0) MONCHL(OFF)
CURRENT RQMNAME( )
STATUS(RETRYING) SUBSTATE(MQGET)
XBATCHSZ(0,0) XMITQ(HERE.TO.THERE.TWO.XMIT)
XQTIME(0,1)
5 MQSC commands read.
No commands have a syntax error.
All valid MQSC commands were processed.
"""
section = parse_info(lines, chr(10))
parsed = parse_ibm_mq_channels(section)
assert 2 + 1 == len(parsed)
attrs = parsed["MY.TEST"]
assert attrs["STATUS"] == "RUNNING"
assert attrs["NOW"] is not None
attrs = parsed["MY.TEST:HERE.TO.THERE.TWO"]
assert attrs["CHLTYPE"] == "SDR"
assert attrs["STATUS"] == "RETRYING"
assert attrs["CONNAME"] == "55.888.222.333(1414),22,333.444.555(1414)"
assert attrs["MONCHL"] == "OFF"
attrs = parsed["MY.TEST:HERE.TO.THERE.TWO"]
assert attrs["CHLTYPE"] == "SDR"
assert attrs["STATUS"] == "RETRYING"
assert attrs["CONNAME"] == "55.888.222.333(1414),22,333.444.555(1414)"
assert attrs["MONCHL"] == "OFF"
def test_parse_svrconn_with_multiple_instances() -> None:
lines = """\
QMNAME(MY.TEST) STATUS(RUNNING) NOW(2020-04-03T17:27:02+0200)
5724-H72 (C) Copyright IBM Corp. 1994, 2015.
Starting MQSC for queue manager MY.TEST.
AMQ8417: Display Channel Status details.
CHANNEL(XXXXXX.IIB.SVRCONN) CHLTYPE(SVRCONN)
CONNAME(10.25.19.182) CURRENT
STATUS(RUNNING) SUBSTATE(RECEIVE)
AMQ8417: Display Channel Status details.
CHANNEL(XXXXXX.IIB.SVRCONN) CHLTYPE(SVRCONN)
CONNAME(10.25.19.183) CURRENT
STATUS(RUNNING) SUBSTATE(RECEIVE)
One MQSC commands read.
No commands have a syntax error.
All valid MQSC commands were processed.
"""
section = parse_info(lines, chr(10))
parsed = parse_ibm_mq_channels(section)
attrs = parsed["MY.TEST:XXXXXX.IIB.SVRCONN"]
assert attrs["CHLTYPE"] == "SVRCONN"
assert attrs["STATUS"] == "RUNNING"
# Last entry of the instances defines the values
assert attrs["CONNAME"] == "10.25.19.183"
def test_discovery_qmgr_not_included() -> None:
check = Check(CHECK_NAME)
parsed = {
"QM1": {"STATUS": "RUNNING"},
"QM1:CHAN1": {"CHLTYPE": "SDR", "STATUS": "RETRYING", "XMITQ": "MY.XMIT.Q"},
"QM1:CHAN2": {"CHLTYPE": "RCVR", "STATUS": "STOPPED"},
"QM1:CHAN3": {"CHLTYPE": "SVRCONN"},
}
discovery = list(check.run_discovery(parsed))
assert len(discovery) == 3
assert ("QM1:CHAN2", {}) in discovery
def test_check() -> None:
check = Check(CHECK_NAME)
params: dict[str, Any] = {}
parsed = {
"QM1": {"STATUS": "RUNNING"},
"QM1:CHAN1": {"CHLTYPE": "SDR", "STATUS": "RETRYING", "XMITQ": "MY.XMIT.Q"},
"QM1:CHAN2": {"CHLTYPE": "RCVR", "STATUS": "STOPPED"},
"QM1:CHAN3": {"CHLTYPE": "SVRCONN"},
}
actual = list(check.run_check("QM1:CHAN1", params, parsed))
expected: list[tuple[int, str, list[Any]]] = [
(1, "Status: RETRYING, Type: SDR, Xmitq: MY.XMIT.Q", [])
]
assert actual == expected
actual = list(check.run_check("QM1:CHAN2", params, parsed))
expected = [(2, "Status: STOPPED, Type: RCVR", [])]
assert actual == expected
actual = list(check.run_check("QM1:CHAN3", params, parsed))
expected = [(0, "Status: INACTIVE, Type: SVRCONN", [])]
assert actual == expected
def test_no_xmit_queue_defined() -> None:
"""
Happened on queue manager MQZZZPPPP and channel FOO.TO.RESA. It
is a misconfiguration on the queue manager, but the monitoring should
not choke on this.
"""
check = Check(CHECK_NAME)
params: dict[str, Any] = {}
parsed = {
"QM1": {"STATUS": "RUNNING"},
"QM1:CHAN1": {"CHLTYPE": "SDR", "STATUS": "RETRYING", "XMITQ": "MY.XMIT.Q"},
"QM1:CHAN2": {"CHLTYPE": "RCVR", "STATUS": "STOPPED"},
"QM1:CHAN3": {"CHLTYPE": "SVRCONN"},
"MQZZZPPPP:FOO.TO.RESA": {"CHLTYPE": "SDR"},
}
actual = list(check.run_check("MQZZZPPPP:FOO.TO.RESA", params, parsed))
expected: list[tuple[int, str, list[Any]]] = [(0, "Status: INACTIVE, Type: SDR", [])]
assert actual == expected
def test_stale_service_for_not_running_qmgr() -> None:
check = Check(CHECK_NAME)
params: dict[str, Any] = {}
parsed = {"QM1": {"STATUS": "ENDED NORMALLY"}}
with pytest.raises(IgnoreResultsError, match=r"Stale because queue manager ENDED NORMALLY"):
list(check.run_check("QM1:CHAN2", params, parsed))
def test_vanished_service_for_running_qmgr() -> None:
check = Check(CHECK_NAME)
params: dict[str, Any] = {}
parsed = {
"QM1": {"STATUS": "RUNNING"},
"QM1:CHAN1": {"CHLTYPE": "SVRCONN"},
}
actual = list(check.run_check("QM1:VANISHED", params, parsed))
assert len(actual) == 0
def test_status_wato_override() -> None:
check = Check(CHECK_NAME)
parsed = {
"QM1": {"STATUS": "RUNNING"},
"QM1:CHAN1": {"CHLTYPE": "SVRCONN", "STATUS": "STOPPED"},
}
# Factory defaults
params: dict[str, Any] = {}
actual = list(check.run_check("QM1:CHAN1", params, parsed))
expected: list[tuple[int, str, list[Any]]] = [(2, "Status: STOPPED, Type: SVRCONN", [])]
assert actual == expected
# Override factory defaults
params = {"mapped_states": [("stopped", 1)]}
actual = list(check.run_check("QM1:CHAN1", params, parsed))
expected = [(1, "Status: STOPPED, Type: SVRCONN", [])]
assert actual == expected
# Override-does-not-match configuration
params = {
"mapped_states": [("retrying", 1)],
"mapped_states_default": 3,
}
actual = list(check.run_check("QM1:CHAN1", params, parsed))
expected = [(3, "Status: STOPPED, Type: SVRCONN", [])]
assert actual == expected