added basic map api routes, and tests

mapping_provider/__init__.py

-"""Initializes the FastAPI application."""
+"""
+Default entry point for the FastAPI application.
+"""
 
 from fastapi import FastAPI
 
-from mapping_provider.api.common import router as version_router
-
+from mapping_provider.api import common, map
 
 def create_app() -> FastAPI:
-    """Create a FastAPI application."""
+    """
+    Creates the FastAPI application instance, with routers attached.
+    """
     app = FastAPI(
         title="Mapping provider",
         description="Mapping provider endpoints for GÉANT maps",
     )
-    app.include_router(version_router)
+    app.include_router(common.router)
+    app.include_router(map.router, prefix='/map')
     return app

mapping_provider/api/map.py

+from importlib.metadata import version
+
+from fastapi import APIRouter
+import jsonschema
+from pydantic import BaseModel
+import requests
+
+router = APIRouter()
+
+
+class Site(BaseModel):
+    latitude: float
+    longitude: float
+    name: str
+
+    @classmethod
+    def from_inprov_site(cls, site: dict) -> 'Site':
+        return cls(
+            latitude=site['latitude'],
+            longitude=site['longitude'],
+            name=site['name']
+        )
+
+class SiteList(BaseModel):
+    sites: list[Site]
+
+class Router(BaseModel):
+    fqdn: str
+    site: str
+
+    @classmethod
+    def from_inprov_router(cls, router: dict) -> 'Router':
+        return cls(
+            fqdn = router['fqdn'],
+            site = router['site']
+        )
+
+class RouterList(BaseModel):
+    routers: list[Router]
+
+
+class Endpoint(BaseModel):
+    hostname: str
+    interface: str
+
+    @classmethod
+    def from_inprov_endpoint(cls, endpoint: dict) -> 'Endpoint':
+        return cls(
+            hostname = endpoint['hostname'],
+            interface = endpoint['interface'],
+        )
+
+class Overlays(BaseModel):
+    speed: int
+
+    @classmethod
+    def from_inprov_overlays(cls, overlays: dict) -> 'Overlays':
+        return cls(
+            speed = overlays['speed'],
+        )
+
+class Service(BaseModel):
+    sid: str
+    name: str
+    type: str
+    endpoints: list[Endpoint]
+    overlays: Overlays
+
+    @classmethod
+    def from_inprov_service(cls, service: dict) -> 'Service':
+        return cls(
+            sid = service['sid'],
+            name = service['name'],
+            type = service['type'],
+            endpoints = list(map(Endpoint.from_inprov_endpoint, service['endpoints'])),
+            overlays = Overlays.from_inprov_overlays(service['overlays']),
+        )
+
+class ServiceList(BaseModel):
+    services: list[Service]
+
+
+INPROV_SITE_LIST_SCHEMA = {
+    '$schema': 'https://json-schema.org/draft/2020-12/schema',
+
+    'definitions': {
+        'site': {
+            'type': 'object',
+            'properties': {
+                'name': {'type': 'string'},
+                'latitude': {'type': 'number'},
+                'longitude': {'type': 'number'},
+            },
+            'required': ['name', 'latitude', 'longitude'],
+            'additionalProperties': True,
+        },
+    },
+
+    'type': 'array',
+    'items': {'$ref': '#/definitions/site'}
+}
+
+INPROV_ROUTER_LIST_SCHEMA = {
+    '$schema': 'https://json-schema.org/draft/2020-12/schema',
+
+    'definitions': {
+        'router': {
+            'type': 'object',
+            'properties': {
+                'fqdn': {'type': 'string'},
+                'site': {'type': 'string'},
+            },
+            'required': ['fqdn', 'site'],
+        },
+    },
+
+    'type': 'array',
+    'items': {'$ref': '#/definitions/router'}
+}
+
+INPROV_SERVICE_LIST_SCHEMA = {
+    '$schema': 'https://json-schema.org/draft/2020-12/schema',
+
+    'definitions': {
+        'endpoint': {
+            'type': 'object',
+            'properties': {
+                'hostname': {'type': 'string'},
+                'interface': {'type': 'string'},
+            },
+        },
+        'service': {
+            'type': 'object',
+            'properties': {
+                'sid': {'type': 'string'},
+                'name': {'type': 'string'},
+                'type': {'type': 'string'},
+                'endpoints': {
+                    'type': 'array',
+                    'items': {'$ref': '#/definitions/endpoint'},
+                    'minItems': 1,
+                },
+                'overlays': {
+                    'type': 'object', 'properties': {
+                        'speed': {'type': 'number'},
+                    },
+                    'required': ['speed'],
+                },
+            },
+            'required': ['sid', 'name', 'type', 'endpoints', 'overlays'],
+        },
+    },
+
+    'type': 'array',
+    'items': {'$ref': '#/definitions/service'}
+}
+
+@router.get("/sites")
+def get_sites() -> SiteList:
+    """
+    handler for /sites
+    """
+
+    # TODO: catch/handle the usual exceptions
+
+    rv = requests.get(
+        'https://test-inprov01.geant.org/map/sites',
+        headers={'Accept': 'application/json'})
+    rv.raise_for_status()
+    site_list_json = rv.json()
+    jsonschema.validate(site_list_json, INPROV_SITE_LIST_SCHEMA)
+
+    return SiteList(sites=map(Site.from_inprov_site, site_list_json))
+
+
+@router.get("/routers")
+def get_routers() -> RouterList:
+    """
+    handler for /sites
+    """
+
+    # TODO: catch/handle the usual exceptions
+
+    rv = requests.get(
+        'https://test-inprov01.geant.org/map/routers',
+        headers={'Accept': 'application/json'})
+    rv.raise_for_status()
+    router_list_json = rv.json()
+    jsonschema.validate(router_list_json, INPROV_ROUTER_LIST_SCHEMA)
+
+    return RouterList(routers=map(Router.from_inprov_router, router_list_json))
+
+
+@router.get("/trunks")
+def get_trunks() -> ServiceList:
+    """
+    handler for /trunks
+    """
+
+    # TODO: catch/handle the usual exceptions
+
+    rv = requests.get(
+        'https://test-inprov01.geant.org/map/services/IP TRUNK',
+        headers={'Accept': 'application/json'})
+    rv.raise_for_status()
+    service_list_json = rv.json()
+    jsonschema.validate(service_list_json, INPROV_SERVICE_LIST_SCHEMA)
+
+    return ServiceList(services=map(Service.from_inprov_service, service_list_json))
+

test/test_map_endpoints.py

+import responses
+
+from mapping_provider.api.map import SiteList, RouterList, ServiceList
+
+
+# @responses.activate
+# def test_inventory_uri_validation():
+#     responses.add(
+#         method=responses.GET, url=re.compile(r".*/version$"), json={"api": "0.9"}
+#     )
+#     assert (
+#         classifier.verify_inventory_provider_uri(None, None, "http://a.b.c:9999")
+#         == "http://a.b.c:9999/"
+#     )
+
+
+def test_get_sites(client):
+    rv = client.get("/map/sites")
+    assert rv.status_code == 200
+    assert rv.json()
+    SiteList.model_validate(rv.json())
+
+
+def test_get_routers(client):
+    rv = client.get("/map/routers")
+    assert rv.status_code == 200
+    assert rv.json()
+    RouterList.model_validate(rv.json())
+
+
+def test_get_trunks(client):
+    rv = client.get("/map/trunks")
+    assert rv.status_code == 200
+    assert rv.json()
+    ServiceList.model_validate(rv.json())