class implementation_check_kind_http_request(interface_check_kind): ''' [implementation] ''' def parameters_schema(self): return { "type": "object", "additionalProperties": False, "properties": { "request": { "type": "object", "additionalProperties": False, "properties": { "target": { "description": "URL", "type": "string" }, "method": { "type": "string", "enum": [ "GET", "POST" ], "default": "GET" } }, "required": [ "target" ] }, "timeout": { "description": "maximum allowed execution time in seconds", "type": "float", "default": 5.0 }, "follow_redirects": { "description": "whether redirect instructions in responses shall be followend instead of being exposed as result", "type": "boolean", "default": False }, "response": { "type": "object", "additionalProperties": False, "properties": { "status_code": { "description": "checks whether the response status code is this", "type": ["null", "integer"], "default": 200 }, "headers": { "description": "conjunctively checks header key-value pairs", "type": "object", "additionalProperties": { "description": "header value", "type": "string" }, "properties": { }, "required": [ ], "default": {} }, "body_part": { "description": "checks whether the response body contains this string", "type": "string" } }, "required": [ ] }, "critical": { "description": "whether a violation of this check shall be leveled as critical instead of concerning", "type": "boolean", "default": True }, "strict": { "deprecated": True, "description": "alias for 'critical'", "type": "boolean", "default": True }, }, "required": [ "request", ] } ''' [implementation] ''' def normalize_order_node(self, node): version = ( "v1" if (not ("critical" in node)) else "v2" ) if (version == "v1"): node_ = dict_merge( { "request": { "method": "GET" }, "timeout": 5.0, "follow_redirects": False, "response": { "status_code": 200 }, "strict": True, }, node, True ) allowed_methods = set(["GET", "POST"]) if (node_["request"]["method"] not in allowed_methods): raise ValueError("invalid HTTP request method: %s" % node_["request"]["method"]) else: return { "request": node_["request"], "timeout": node_["timeout"], "follow_redirects": node_["follow_redirects"], "response": node_["response"], "critical": node_["strict"], } elif (version == "v2"): node_ = dict_merge( { "request": { "method": "GET" }, "timeout": 5.0, "follow_redirects": False, "response": { "status_code": 200 }, "critical": True, }, node, True ) allowed_methods = set(["GET", "POST"]) if (node_["request"]["method"] not in allowed_methods): raise ValueError("invalid HTTP request method: %s" % node_["request"]["method"]) else: return node_ else: raise ValueError("unhandled") ''' [implementation] ''' def run(self, parameters): if (parameters["request"]["method"] == "GET"): try: response = _requests.get( parameters["request"]["target"], timeout = parameters["timeout"], allow_redirects = parameters["follow_redirects"] ) error = None except Exception as error_: error = error_ response = None elif (parameters["request"]["method"] == "POST"): try: response = _requests.post( parameters["request"]["target"], timeout = parameters["timeout"], allow_redirects = parameters["follow_redirects"] ) error = None except Exception as error_: error = error_ response = None else: raise ValueError("impossible") if (response is None): return { "condition": ( enum_condition.critical if parameters["strict"] else enum_condition.concerning ), "info": { "request": parameters["request"], "faults": [ translation_get("checks.http_request.request_failed"), ], } } else: faults = [] for (key, value, ) in parameters["response"].items(): if (key == "status_code"): if ((value is None) or (response.status_code == value)): pass else: faults.append( translation_get( "checks.http_request.status_code_mismatch", { "status_code_actual": ("%u" % response.status_code), "status_code_expected": ("%u" % value), } ) ) elif (key == "headers"): for (header_key, header_value, ) in value.items(): if (not (header_key in response.headers)): faults.append( translation_get( "checks.http_request.header_missing", { "key": header_key, "value_expected": header_value, } ) ) else: if (not (response.headers[header_key] == header_value)): faults.append( translation_get( "checks.http_request.header_value_mismatch", { "key": header_key, "value_actual": response.headers[header_key], "value_expected": header_value, } ) ) else: pass elif (key == "body_part"): if (response.text.find(value) >= 0): pass else: faults.append( translation_get( "checks.http_request.body_misses_part", { "part": value, } ) ) else: raise ValueError("unhandled") return { "condition": ( enum_condition.ok if (len(faults) <= 0) else ( enum_condition.critical if parameters["strict"] else enum_condition.concerning ) ), "info": { "request": parameters["request"], "response": { "status_code": response.status_code, "headers": dict(map(lambda pair: pair, response.headers.items())), # "body": response.text, }, "faults": faults, } }