class implementation_check_kind_http_request(interface_check_kind): ''' [implementation] ''' def normalize_conf_node(self, node): return dict_merge( { "request": { "method": "GET" }, "response": { "status_code": 200 }, "as_warning": False, }, node, True ) ''' [implementation] ''' def run(self, parameters): if (parameters["request"]["method"] == "GET"): method_handled = True try: response = _requests.get( parameters["request"]["target"] ) error = None except Exception as error_: error = error_ response = None elif (parameters["request"]["method"] == "POST"): method_handled = True try: response = _requests.post( parameters["request"]["target"] ) error = None except Exception as error_: error = error_ response = None else: method_handled = False response = None if (not method_handled): return { "condition": enum_condition.unknown, "output": ("invalid HTTP request method: %s" % parameters["request"]["method"]) } else: if (response is None): return { "condition": ( enum_condition.warning if parameters["as_warning"] else enum_condition.critical ), "output": "HTTP request failed", } else: lines = [] for (key, value, ) in parameters["response"].items(): if (key == "status_code"): if ((value is None) or (response.status_code == value)): pass else: lines.append( string_coin( "actual status code {{status_code_actual}} does not match expected value {{status_code_expected}}", { "status_code_actual": ("%u" % response.status_code), "status_code_expected": ("%u" % value), } ) ) elif (key == "headers"): for (header_key, header_value, ) in value.items(): if (response.headers[header_key] == header_value): pass else: lines.append( string_coin( "actual header value for key {{key}} is {{value_actual}} and does not match the expected value {{value_expected}}", { "key": header_key, "value_actual": response.headers[header_key], "value_expected": header_value, } ) ) elif (key == "body_part"): if (response.text.find(value) >= 0): pass else: lines.append( string_coin( "body does not contain the expected part '{{part}}'", { "part": value, } ) ) else: raise ValueError("unhandled ") return { "condition": ( enum_condition.ok if (len(lines) <= 0) else ( enum_condition.warning if parameters["as_warning"] else enum_condition.critical ) ), "output": "\n".join(lines), }