diff --git a/mapping_provider/api/map.py b/mapping_provider/api/map.py index 54fdba9b8b3f03bee869be61c44731966415784c..b8bc36b921069f51f3ca75e7bc954f13336d8fd9 100644 --- a/mapping_provider/api/map.py +++ b/mapping_provider/api/map.py @@ -146,3 +146,17 @@ def get_trunks() -> services.ServiceList: handler for /trunks, same as /services/IP TRUNK """ return get_services(service_type='IP TRUNK') + + +@router.get("/map-info") +@router.get("/map-info/{service_type}") +def get_map_info(service_type: str | None = None) -> services.MapInfo: + """ + handler for /map-info + """ + pop_list = get_pops() + pops_dict = {_p.name: _p.abbreviation for _p in pop_list.pops} + return_value = services.build_map_info(pop_abbrevs=pops_dict, service_type=service_type) + if not return_value.nodes: + raise HTTPException(status_code=404, detail=f'unrecognized service type: {service_type}') + return return_value diff --git a/mapping_provider/backends/services.py b/mapping_provider/backends/services.py index 7406a0ebb29dbc150a2fd2297eb978aa22cd1a22..a19bddc5e477ffaf217806ab4a4c002644bfb509 100644 --- a/mapping_provider/backends/services.py +++ b/mapping_provider/backends/services.py @@ -38,6 +38,11 @@ class ServiceList(BaseModel): services: list[Service] +class MapInfo(BaseModel): + nodes: list[str] + edges: list[str] + + def endpoint_equipment(endpoint: dict[str, Any]) -> str: """ convert the correlator router hostname or optical equipment name @@ -145,3 +150,45 @@ def build_service_info_list(service_type: str | None = None) -> ServiceList: return a list of mappable info about all operational services """ return ServiceList(services=list(_services(service_type))) + + +def _service_label_in_map(service: Service) -> str | None: + """ + return a label for the service in the map + """ + if not service.pops: + return None + + pops = sorted(service.pops) + label_components = pops + [service.scid[:8].upper()] + return '-'.join(label_components) + + +def build_map_info(pop_abbrevs: dict[str, str], service_type: str | None = None) -> MapInfo: + """ + return a list of mappable info about operational services + """ + nodes = set() + edges = set() + + def _get_service_pops(svc: Service) -> Generator[str]: + for _p in svc.pops: + abbrev = pop_abbrevs.get(_p) + if abbrev: + yield abbrev + + for _s in _services(service_type): + + _pops = set(_get_service_pops(_s)) + if not _pops: + logger.warning(f'service {_s.sid} has no POPs') + continue + + nodes |= set(_pops) + + _lc = list(_pops) + [_s.scid[:8].upper()] + edges.add('-'.join(_lc)) + + return MapInfo( + nodes=sorted(nodes), + edges=sorted(edges)) diff --git a/test/test_map_endpoints.py b/test/test_map_endpoints.py index c3dbaba94c42eed1fdaf0c348f7aa657bda64b87..457eb3b7ad605ab959d1155c21c6f38c1bc3cc95 100644 --- a/test/test_map_endpoints.py +++ b/test/test_map_endpoints.py @@ -4,20 +4,22 @@ import pytest import responses from mapping_provider.api.map import EquipmentList, PopList -from mapping_provider.backends.services import ServiceList +from mapping_provider.backends.services import MapInfo, ServiceList from .common import load_test_data -@responses.activate -def test_get_pops(client): - +def _add_pops_response(): responses.add( method=responses.GET, url=re.compile(r'.*/map/pops$'), json=load_test_data('inprov-pops.json') ) +@responses.activate +def test_get_pops(client): + _add_pops_response() + rv = client.get("/map/pops") assert rv.status_code == 200 pop_list = PopList.model_validate(rv.json()) @@ -86,6 +88,7 @@ def test_get_all_services(client): service_list = ServiceList.model_validate(rv.json()) assert service_list.services, 'test data should not be empty' + @responses.activate def test_get_trunks(client): rv = client.get("/map/trunks") @@ -95,3 +98,20 @@ def test_get_trunks(client): assert all(s.type == 'IP TRUNK' for s in service_list.services) +@responses.activate +def test_get_trunk_map_info(client): + _add_pops_response() + + rv = client.get("/map/map-info/IP TRUNK") + assert rv.status_code == 200 + map_info = MapInfo.model_validate(rv.json()) + assert map_info.nodes, 'test data should not be empty' + assert map_info.edges, 'test data should not be empty' + + +@responses.activate +def test_get_unknown_service_type_map_info(client): + _add_pops_response() + rv = client.get("/map/map-info/BOGUS_SERVICE_TYPE") + assert rv.status_code == 404 +