Commit d85f2f71 authored by kaiyou's avatar kaiyou

Add an http check to grab data from an xpath request

parent bdd0b6d3
from amonit import util
import requests
import requests_html
@util.wrap
def simple(context, url, expected_code=None):
def get(context, url, expected_code=200, expected_string=None):
""" Run a GET http request and test the HTTP response code
"""
req = requests.get(url)
if expected_code != req.status_code:
up = False
message = "the page returned code {}".format(req.status_code)
elif req and expected_string:
up = False
message = "string {} not found in the response".format(expected_string)
else:
up = True
message = ""
return {
"up": req.status_code == (expected_code or 200),
"up": up,
"code": req.status_code,
"message": "the page returned code {}".format(req.status_code)
"message": message,
"delay": req.elapsed.microseconds
}
@util.wrap
def get_xpath(context, url, query, result_type="str", destination="message"):
""" Run a GET http requests and use xpath to extract a result
"""
cast = {"str": str, "int": int, "float": float, "bool": bool}[result_type]
session = requests_html.HTMLSession()
req = session.get(url)
result = req.html.xpath(query, first=True)
return {
"up": True,
destination: None if result is None else cast(result)
}
......@@ -61,6 +61,7 @@ def check_run(checkid, function, context, args):
context, result = scheduler.hook("postcheck", context, result)
scheduler.state[checkid] = context
scheduler.notify(checkid, context)
return result
@celery.current_app.task
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment