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))