testing.py

import json
import os

from flask import Blueprint, Response, jsonify, current_app
from lxml import etree

from inventory_provider import juniper
from inventory_provider.routes import common
from inventory_provider.tasks import worker
from inventory_provider.tasks import common as worker_common

routes = Blueprint("inventory-data-testing-support-routes", __name__)


@routes.route("flushdb", methods=['GET', 'POST'])
def flushdb():
    common.get_current_redis().flushdb()
    return Response('OK')


# IMS routes


@routes.route("update-interfaces-to-services", methods=['GET', 'POST'])
def update_interfaces_to_services_ims():
    worker.update_interfaces_to_services.delay(use_current=True)
    return Response('OK')


@routes.route("update-fibre-spans", methods=['GET', 'POST'])
def update_fibre_spans_ims():
    worker.update_fibre_spans.delay(use_current=True)
    return Response('OK')


@routes.route("update-circuit-hierarchy", methods=['GET', 'POST'])
def update_circuit_hierarchy_ims():
    worker.update_circuit_hierarchy.delay(use_current=True)
    return Response('OK')


@routes.route("update-equipment-locations", methods=['GET', 'POST'])
def update_equipment_locations_ims():
    worker.update_equipment_locations.delay(use_current=True)
    return Response('OK')


@routes.route("update-lg-routers", methods=['GET', 'POST'])
def update_lg_routers_ims():
    worker.update_lg_routers.delay(use_current=True)
    return Response('OK')

# End of IMS routes


@routes.route("juniper-server-addresses", methods=['GET', 'POST'])
@common.require_accepts_json
def juniper_addresses():
    # TODO: this route (and corant, infinera routes) can be removed
    r = common.get_current_redis()
    routers = r.get('netdash')
    assert routers  # sanity: value shouldn't be empty
    routers = json.loads(routers.decode('utf-8'))
    return jsonify(routers)


@routes.route("bgp/<hostname>", methods=['GET', 'POST'])
@common.require_accepts_json
def bgp_configs(hostname):
    r = common.get_current_redis()
    netconf_string = r.get('netconf:' + hostname)
    if not netconf_string:
        return Response(
            response="no available info for '%s'" % hostname,
            status=404,
            mimetype="text/html")

    routes = list(juniper.all_bgp_peers(
        etree.XML(netconf_string.decode('utf-8'))))
    if not routes:
        return Response(
            response="no interfaces found for '%s'" % hostname,
            status=404,
            mimetype="text/html")

    return jsonify(routes)


@routes.route("snmp/<hostname>", methods=['GET', 'POST'])
@common.require_accepts_json
def snmp_ids(hostname):
    r = common.get_next_redis()
    ifc_data_string = r.get('snmp-interfaces:' + hostname)
    ifc_data = json.loads(ifc_data_string.decode('utf-8'))
    return jsonify(ifc_data)


@routes.route("netconf/<hostname>", methods=['GET', 'POST'])
def get_netconf(hostname):
    config = current_app.config["INVENTORY_PROVIDER_CONFIG"]
    try:
        netconf_doc = juniper.load_config(
            hostname, config["ssh"], validate=False)
        msg = etree.tostring(netconf_doc, encoding='unicode')
    except (ConnectionError, juniper.NetconfHandlingError) as e:
        msg = f'error loading netconf data from {hostname}\n{e}'
    return msg


@routes.route("routers-list", methods=['GET', 'POST'])
def routers_from_config_dir():
    with open(os.environ['FAKE_ROUTERS_FILE']) as f:
        return Response(f.read(), mimetype="text.plain")


@routes.route("latchdb", methods=['GET', 'POST'])
def latch_db():
    config = current_app.config["INVENTORY_PROVIDER_CONFIG"]
    worker_common.latch_db(config)
    r = worker_common.get_current_redis(config)
    return jsonify(worker_common.get_latch(r))