diff --git a/inventory_provider/routes/neteng.py b/inventory_provider/routes/neteng.py index 29f8abcd9c963c06ac40fed5063d32f401435a28..867f7715d4ce722eeb0eadc414889599a37456fb 100644 --- a/inventory_provider/routes/neteng.py +++ b/inventory_provider/routes/neteng.py @@ -15,6 +15,7 @@ These endpoints are intended for use by neteng tools. """ import json import logging +import re import threading from flask import Blueprint, Response, jsonify @@ -25,6 +26,12 @@ routes = Blueprint('neteng-query-routes', __name__) logger = logging.getLogger(__name__) _subnet_lookup_semaphore = threading.Semaphore() +STRING_LIST_SCHEMA = { + '$schema': 'http://json-schema.org/draft-07/schema#', + 'type': 'array', + 'items': {'type': 'string'} +} + @routes.after_request def after_request(resp): @@ -66,3 +73,30 @@ def get_location(equipment): mimetype='text/html') return jsonify(value[0]) + + +@routes.route('/pops', methods=['GET', 'POST']) +@common.require_accepts_json +def get_pops(): + """ + Handler for `/neteng/pops` + + This method will return a list of defined pop + names. Elements from this list can be used + with `/neteng/pop`. + + .. asjson:: + inventory_provider.db.ims_data.STRING_LIST_SCHEMA + + :return: as above + """ + + def _pops(): + r = common.get_current_redis() + for k in r.scan_iter('ims:pop:*', count=1000): + k = k.decode('utf-8') + m = re.match('^ims:pop:(.+)$', k) + yield m.group(1) + + return jsonify(sorted(list(_pops()))) + diff --git a/test/test_neteng_routes.py b/test/test_neteng_routes.py index 1e95322a3be4cd43a83caeea12c2a9f6d35e034e..23a7bba78bbb960a69fe63a240e7302e1cc588b2 100644 --- a/test/test_neteng_routes.py +++ b/test/test_neteng_routes.py @@ -2,7 +2,9 @@ import json import jsonschema import pytest -from inventory_provider.db.ims_data import NODE_LOCATION_SCHEMA +from inventory_provider.db.ims_data \ + import NODE_LOCATION_SCHEMA, POP_LOCATION_SCHEMA +from inventory_provider.routes.neteng import STRING_LIST_SCHEMA @pytest.mark.parametrize('equipment_name', [ @@ -25,3 +27,13 @@ def test_location_not_found(client, mocked_redis): '/neteng/location/BOGUS.EQUIPMENT.NAME', headers={'Accept': ['application/json']}) assert rv.status_code == 404 + + +def test_get_pops(client, mocked_redis): + rv = client.get( + f'/neteng/pops', + headers={'Accept': ['application/json']}) + assert rv.status_code == 200 + jsonschema.validate( + json.loads(rv.data.decode('utf-8')), + STRING_LIST_SCHEMA)