Finished feature all-services-endpoint.

mapping_provider/api/map.py

@@ -2,7 +2,7 @@ from typing import Any
 
 import jsonschema
 import requests
-from fastapi import APIRouter
+from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
 from mapping_provider import config
@@ -128,9 +128,21 @@ def get_equipment() -> EquipmentList:
     return EquipmentList(equipment=list(map(_make_equipment, equipment_list_obj)))
 
 
+@router.get("/services")
+@router.get("/services/{service_type}")
+def get_services(service_type: str | None = None) -> services.ServiceList:
+    """
+    handler for /trunks
+    """
+    return_value = services.build_service_info_list(service_type=service_type)
+    if not return_value.services:
+        raise HTTPException(status_code=404, detail=f'unrecognized service type: {service_type}')
+    return return_value
+
+
 @router.get("/trunks")
 def get_trunks() -> services.ServiceList:
     """
-    handler for /trunks
+    handler for /trunks, same as /services/IP TRUNK
     """
-    return services.build_service_info_list(service_type='IP TRUNK')
+    return get_services(service_type='IP TRUNK')

mapping_provider/backends/services.py

-import functools
 import logging
 import re
 from collections.abc import Generator
@@ -30,6 +29,8 @@ class Service(BaseModel):
     name: str
     type: str
     pops: list[str]
+    # TODO: temporarily removed for simplicity (first map POC is only POP-based)
+    # equipment: list[str]
     overlays: Overlays
 
 
@@ -37,8 +38,11 @@ class ServiceList(BaseModel):
     services: list[Service]
 
 
-def endpoint_to_pop(endpoint: dict[str, Any], inprov_equipment_dict: dict[str, dict[str, Any]]) -> str:
-    
+def endpoint_equipment(endpoint: dict[str, Any]) -> str:
+    """
+    convert the correlator router hostname or optical equipment name
+    to the inventory equipment format
+    """
     def _hostname_to_equipment(_hn: str) -> str:
         m = re.match(r'^(.+)\.geant\.net$', _hn)
         if not m:
@@ -47,21 +51,14 @@ def endpoint_to_pop(endpoint: dict[str, Any], inprov_equipment_dict: dict[str, d
         return m.group(1).upper()
 
     if 'hostname' in endpoint:
-        eq_name = _hostname_to_equipment(endpoint['hostname'])
+        return _hostname_to_equipment(endpoint['hostname'])
     elif 'equipment' in endpoint:
-        eq_name = endpoint['equipment']
-    else:
-        # should already be validated
-        raise AssertionError(f'no equipment or hostname in endpoint: {endpoint}')
-
-    if eq_name not in inprov_equipment_dict:
-        # TODO: is this really possible if all data is read from IMS at the same time?
-        logger.error(f'unknown endpoint equipment: {eq_name}')
-        return '?'
+        _name = endpoint['equipment']
+        assert isinstance(_name, str)  # mypy noise
+        return _name
 
-    pop_name = inprov_equipment_dict[eq_name]['pop']
-    assert isinstance(pop_name, str)  # mypy noise
-    return pop_name
+    # should already be validated
+    raise AssertionError(f'no equipment or hostname in endpoint: {endpoint}')
 
 
 def _services(service_type: str | None = None) -> Generator[Service]:
@@ -92,7 +89,15 @@ def _services(service_type: str | None = None) -> Generator[Service]:
     brian_scid_rates = {r['scid']: r['values'] for r in brian_rates}
 
     equipment_dict = {_x['name']: _x for _x in equipment_list}
-    _endpoint_to_pop = functools.partial(endpoint_to_pop, inprov_equipment_dict=equipment_dict)
+
+    def _get_equipment_pop(equipment_name: str) -> str:
+        if equipment_name not in equipment_dict:
+            # TODO: is this really possible if all data is read from IMS at the same time?
+            logger.error(f'unknown endpoint equipment: {equipment_name}')
+            return '?'
+        _pop_name = equipment_dict[equipment_name]['pop']
+        assert isinstance(_pop_name, str)  # mypy noise
+        return _pop_name
 
     for _s in scid_current:
 
@@ -102,23 +107,24 @@ def _services(service_type: str | None = None) -> Generator[Service]:
         if service_type and _s['service_type'] != service_type:
             continue
 
-        pops = sorted(set(map(_endpoint_to_pop, _s['endpoints'])))
+        equipment = sorted(set(map(endpoint_equipment, _s['endpoints'])))
+        pops = sorted(set(map(_get_equipment_pop, equipment)))
         
         rates = brian_scid_rates.get(_s['scid'], {})
         overlays = Overlays(
             speed = _s['speed'],
             up = _s['sid'] not in down_service_sids,
             latest = BitRates(
-                egress = rates['latest']['egress'],
-                ingress = rates['latest']['ingress'],
+                egress = rates.get('latest', {}).get('egress'),
+                ingress = rates.get('latest', {}).get('ingress'),
             ),
             mean = BitRates(
-                egress = rates['mean']['egress'],
-                ingress = rates['mean']['ingress'],
+                egress = rates.get('mean', {}).get('egress'),
+                ingress = rates.get('mean', {}).get('ingress'),
             ),
             max = BitRates(
-                egress = rates['max']['egress'],
-                ingress = rates['max']['ingress'],
+                egress = rates.get('max', {}).get('egress'),
+                ingress = rates.get('max', {}).get('ingress'),
             ),
         )
 
@@ -128,6 +134,8 @@ def _services(service_type: str | None = None) -> Generator[Service]:
             name = _s['name'],
             type = _s['service_type'],
             pops = pops,
+            # TODO: temporarily removed for simplicity (first map POC is only POP-based)
+            # equipment = equipment,
             overlays = overlays,
         )
 

test/test_map_endpoints.py

 import re
 
+import pytest
 import responses
 
 from mapping_provider.api.map import EquipmentList, PopList
@@ -38,9 +39,59 @@ def test_get_equipment(client):
     assert equipment_list.equipment, 'test data should not be empty'
 
 
+
+
+@responses.activate
+@pytest.mark.parametrize('service_type', [
+    'IP PEERING - R&E',
+    'GEANT SPECTRUM SERVICE',
+    'L3-VPN',
+    'OOB IP LINK',
+    'GEANT OPEN CROSS CONNECT',
+    'GEANT - GBS',
+    'GWS - INDIRECT',
+    'GEANT IP',
+    'POP LAN LINK',
+    'IP PEERING - NON R&E (PUBLIC)',
+    'IP TRUNK',
+    'GEANT PLUS',
+    'L2SERVICES',
+    'ETHERNET',
+    'EUMETSAT TERRESTRIAL',
+    'IP PEERING - NON R&E (PRIVATE)',
+    'EXPRESS ROUTE',
+    'CBL1',
+    'GWS - UPSTREAM',
+    'GEANT PEERING',
+    'SERVER LINK',
+    'GEANT MANAGED WAVELENGTH SERVICE',
+    'CORPORATE',
+    'EUMETSAT GRE'])
+def test_get_services(client, service_type):
+    rv = client.get(f"/map/services/{service_type}")
+    assert rv.status_code == 200
+    service_list = ServiceList.model_validate(rv.json())
+    assert service_list.services, 'test data should not be empty'
+    assert all(s.type == service_type for s in service_list.services)
+
+
+def test_get_unknown_service_type(client):
+    rv = client.get("/map/services/BOGUS_SERVICE_TYPE")
+    assert rv.status_code == 404
+
+
+def test_get_all_services(client):
+    rv = client.get("/map/services")
+    assert rv.status_code == 200
+    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")
     assert rv.status_code == 200
     service_list = ServiceList.model_validate(rv.json())
     assert service_list.services, 'test data should not be empty'
+    assert all(s.type == 'IP TRUNK' for s in service_list.services)
+
+