dispatcher.py 文件源码

def test_source(name):
    orm_session = model.Session()
    source = orm_session.query(model.Source).filter(model.Source.source_name==name).first()
    orm_session.close()
    if source is None:
        error_msg = "Source " + name + " not found!"
        if json_check():
            return json_error(error_msg, 404)
        flash(error_msg, "alert-danger")
        return redirect("/sources")
    recv = receiver.OpennmsReceiver(source)
    test_result = recv.test_connection()
    result = {}
    result["result_state"] = "failed"
    result["result_msg"] = ""
    if test_result == 200:
        result["result_state"] = "success"
        result["result_msg"] = "Test of source " + name + " successful: HTTP/" + str(test_result)
        flash(result["result_msg"], "alert-success")
    elif test_result == -1:
        result["result_msg"] = "Test of source " + name + " failed: Error connecting to server"
        flash(result["result_msg"], "alert-danger")
    else:
        result["result_msg"] = "Test of source " + name + " failed: HTTP/" + str(test_result)
        flash(result["result_msg"], "alert-danger")
    if json_check():
        return jsonify(result)
    return redirect("/sources")