Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision

Target

Select target project
  • geant-swd/next-gen-map/mapping-provider
1 result
Select Git revision
Show changes
Commits on Source (11)
Showing
with 1340 additions and 98 deletions
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
build/
dist/
*.egg-info/
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Virtual environments
venv/
.env
.venv/
# PyInstaller
*.manifest
*.spec
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.tox
bom.json
coverage.xml
.cache
.pytest_cache/
# Sphinx documentation
docs/_build/
# Editor / OS junk
.coverage
htmlcov
*.egg-info
__pycache__
.vscode
docs/build
.DS_Store
Thumbs.db
.idea/
.vscode/
*.swp
*.swo
\ No newline at end of file
# drawio tmp files
*.bkp
......@@ -7,6 +7,7 @@ from datetime import datetime
import json
import os
import sys
import tempfile
sys.path.insert(
0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
......@@ -28,7 +29,7 @@ release = '0.0'
extensions = [
"sphinx_rtd_theme",
# "sphinx.ext.autodoc",
"sphinx.ext.autodoc",
"sphinx.ext.coverage",
"sphinxcontrib.plantuml",
"sphinxcontrib.openapi",
......@@ -53,7 +54,19 @@ html_theme = 'sphinx_rtd_theme'
html_static_path = ['_static']
api_schema = create_app().openapi()
# we need a minimal/parseable config file in order to
# start the server and dump the schema
with tempfile.NamedTemporaryFile(delete=False) as f:
bogus_config = {'inventory': 'http://bogus'}
with open(f.name, 'w') as f:
json.dump(bogus_config, f)
f.flush()
os.environ['SETTINGS_FILENAME'] = f.name
api_schema = create_app().openapi()
openapi_filename = os.path.join(os.path.dirname(__file__), "openapi.json")
with open(openapi_filename, 'w') as f:
json.dump(api_schema, f, indent=4)
mapping_provider package
=========================
Submodules
----------
mapping\_provider.main module
-----------------------------
.. automodule:: mapping_provider.main
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
.. automodule:: mapping_provider
:members:
:undoc-members:
:show-inheritance:
mapping_provider
================
.. toctree::
:maxdepth: 4
mapping_provider
"""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 import config, environment
from mapping_provider.api import common, map
def create_app() -> FastAPI:
"""Create a FastAPI application."""
"""
Creates the FastAPI application instance, with routers attached.
"""
environment.setup_logging()
app_config = config.load()
if app_config.sentry:
environment.setup_sentry(app_config.sentry)
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
from importlib.metadata import PackageNotFoundError, version
from importlib.metadata import version
from fastapi import APIRouter
from pydantic import BaseModel
router = APIRouter()
class Version(BaseModel):
module: str
api: str
@router.get("/version")
def get_version() -> dict[str, str]:
"""Get the version of the package."""
try:
return {"version": version("mapping_provider")}
except PackageNotFoundError:
return {"version": "unknown"}
def get_version() -> Version:
"""
handler for /version
"""
return Version(
module = version('mapping_provider'),
api = '0.1'
)
from typing import Any
import jsonschema
import requests
from fastapi import APIRouter
from pydantic import BaseModel
from mapping_provider import config
router = APIRouter()
class Site(BaseModel):
latitude: float
longitude: float
name: str
@classmethod
def from_inprov_site(cls, site: dict[str, Any]) -> '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[str, Any]) -> '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[str, Any]) -> 'Endpoint':
return cls(
hostname = endpoint['hostname'],
interface = endpoint['interface'],
)
class Overlays(BaseModel):
speed: int
@classmethod
def from_inprov_overlays(cls, overlays: dict[str, Any]) -> '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[str, Any]) -> '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'}
}
INPROV_API_URL_TODO = 'https://test-inprov01.geant.org'
@router.get("/sites")
def get_sites() -> SiteList:
"""
handler for /sites
"""
# TODO: catch/handle the usual exceptions
app_params = config.load()
rv = requests.get(
f'{app_params.inventory}/map/sites',
headers={'Accept': 'application/json'})
rv.raise_for_status()
site_list_json = rv.json()
jsonschema.validate(site_list_json, INPROV_SITE_LIST_SCHEMA)
rsp_sites = map(Site.from_inprov_site, site_list_json)
return SiteList(sites=list(rsp_sites))
@router.get("/routers")
def get_routers() -> RouterList:
"""
handler for /sites
"""
# TODO: catch/handle the usual exceptions
app_params = config.load()
rv = requests.get(
f'{app_params.inventory}/map/routers',
headers={'Accept': 'application/json'})
rv.raise_for_status()
router_list_json = rv.json()
jsonschema.validate(router_list_json, INPROV_ROUTER_LIST_SCHEMA)
rsp_routers = map(Router.from_inprov_router, router_list_json)
return RouterList(routers=list(rsp_routers))
@router.get("/trunks")
def get_trunks() -> ServiceList:
"""
handler for /trunks
"""
# TODO: catch/handle the usual exceptions
app_params = config.load()
rv = requests.get(
f'{app_params.inventory}/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)
rsp_services = map(Service.from_inprov_service, service_list_json)
return ServiceList(services=list(rsp_services))
import os
from pydantic import BaseModel, Field, HttpUrl
class SentryConfig(BaseModel):
dsn: str
environment: str
level: str = Field(
default='error',
pattern="^(debug|info|warning|error)$")
# @field_validator('level')
# @classmethod
# def validate_level(cls, v):
# if v not in ['debug', 'info', 'warning', 'error']:
# raise ValueError('level must be one of: debug, info, warning, error')
# return v
class Configuration(BaseModel):
sentry: SentryConfig | None = None
inventory: HttpUrl
def load() -> Configuration:
"""
Loads, validates and returns configuration parameters from
the file named in the environment variable 'SETTINGS_FILENAME'.
:return:
"""
assert 'SETTINGS_FILENAME' in os.environ
with open(os.environ['SETTINGS_FILENAME']) as f:
return Configuration.model_validate_json(f.read())
"""
Environment setup
===================
.. autofunction:: mapping_provider.environment.setup_logging
.. autofunction:: mapping_provider.environment.setup_sentry
"""
import json
import logging
import logging.config
import os
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from mapping_provider.config import SentryConfig
import sentry_sdk
from sentry_sdk.integrations.logging import LoggingIntegration
LOGGING_DEFAULT_CONFIG = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'simple': {
'format': '%(asctime)s - %(name)s '
'(%(lineno)d) - %(levelname)s - %(message)s'
}
},
'handlers': {
'console': {
'class': 'logging.StreamHandler',
'level': 'DEBUG',
'formatter': 'simple',
'stream': 'ext://sys.stdout'
}
},
'loggers': {
'mapping_provider': {
'level': 'DEBUG',
'handlers': ['console'],
'propagate': False
}
},
'root': {
'level': 'WARNING',
'handlers': ['console']
}
}
def setup_logging() -> None:
"""
Sets up logging using the configured filename.
if LOGGING_CONFIG is defined in the environment, use this for
the filename, otherwise use LOGGING_DEFAULT_CONFIG
"""
logging_config = LOGGING_DEFAULT_CONFIG
if 'LOGGING_CONFIG' in os.environ:
filename = os.environ['LOGGING_CONFIG']
with open(filename) as f:
logging_config = json.load(f)
logging.config.dictConfig(logging_config)
def setup_sentry(sentry_config_params: 'SentryConfig') -> None:
"""
Sets up the sentry instrumentation based on the Configuration.sentry params.
"""
match str(sentry_config_params.level):
case 'debug':
level = logging.DEBUG
case 'info':
level = logging.INFO
case 'warning':
level = logging.WARNING
case 'error':
level = logging.ERROR
sentry_sdk.init(
dsn=sentry_config_params.dsn,
environment=sentry_config_params.environment,
integrations=[
LoggingIntegration(
level=logging.INFO, # Capture info and above as breadcrumbs
event_level=level # Send records as events
),
],
)
[tool.ruff]
line-length = 120
target-version = "py313"
select = ["E", "F", "I", "B", "UP", "N"]
fixable = ["ALL"]
lint.select = ["E", "F", "I", "B", "UP", "N"]
lint.fixable = ["ALL"]
exclude = ["tests", "docs", "build"]
[tool.mypy]
python_version = "3.13"
strict = true
warn_unused_ignores = true
warn_return_any = true
\ No newline at end of file
warn_return_any = true
......@@ -13,7 +13,10 @@ setup(
python_requires=">=3.10",
install_requires=[
"fastapi",
"uvicorn[standard]"
"uvicorn[standard]",
"requests",
"jsonschema",
"sentry_sdk",
],
long_description=open("README.md", encoding="utf-8").read(),
long_description_content_type="text/markdown",
......
import json
import os
import tempfile
from unittest.mock import patch
import pytest
from fastapi.testclient import TestClient
from mapping_provider import create_app
@pytest.fixture
def dummy_config():
return {
'sentry': {
'dsn': 'https://token@hostname.geant.org:1111/a/b',
'environment': 'unit tests'
},
'inventory': 'https://dummy-hostname.dummy.domain'
}
@pytest.fixture
def dummy_config_filename(dummy_config):
with tempfile.NamedTemporaryFile(delete=False) as f:
f.write(json.dumps(dummy_config).encode('utf-8'))
f.flush()
yield f.name
@pytest.fixture
def client(dummy_config_filename):
os.environ['SETTINGS_FILENAME'] = dummy_config_filename
with patch('sentry_sdk.init') as _mock_sentry_init:
return TestClient(create_app())
[
{
"fqdn": "qfx.par.fr.geant.net",
"site": "PAR"
},
{
"fqdn": "mx2.lis.pt.geant.net",
"site": "LIS"
},
{
"fqdn": "mx1.vie.at.geant.net",
"site": "VIE"
},
{
"fqdn": "qfx.fra.de.geant.net",
"site": "FRA"
},
{
"fqdn": "qfx.lon2.uk.geant.net",
"site": "LON2"
},
{
"fqdn": "mx1.poz.pl.geant.net",
"site": "POZ"
},
{
"fqdn": "rt1.bil.es.geant.net",
"site": "BIL"
},
{
"fqdn": "rt1.chi.md.geant.net",
"site": "CHI"
},
{
"fqdn": "rt0.ath2.gr.geant.net",
"site": "ATH2"
},
{
"fqdn": "rt1.ams.nl.geant.net",
"site": "AMS"
},
{
"fqdn": "rt1.buc.ro.geant.net",
"site": "BUC"
},
{
"fqdn": "rt1.bra.sk.geant.net",
"site": "BRA"
},
{
"fqdn": "rt0.lon.uk.geant.net",
"site": "LON"
},
{
"fqdn": "mx2.zag.hr.geant.net",
"site": "ZAG"
},
{
"fqdn": "srx1.am.office.geant.net",
"site": "AMO"
},
{
"fqdn": "mx1.bud.hu.geant.net",
"site": "BUD"
},
{
"fqdn": "mx1.buc.ro.geant.net",
"site": "BUC"
},
{
"fqdn": "rt0.mil2.it.geant.net",
"site": "MIL2"
},
{
"fqdn": "rt0.mar.fr.geant.net",
"site": "MAR"
},
{
"fqdn": "mx1.lon2.uk.geant.net",
"site": "LON2"
},
{
"fqdn": "mx1.gen.ch.geant.net",
"site": "GEN"
},
{
"fqdn": "mx1.mad.es.geant.net",
"site": "MAD"
},
{
"fqdn": "mx1.par.fr.geant.net",
"site": "PAR"
},
{
"fqdn": "mx1.lon.uk.geant.net",
"site": "LON"
},
{
"fqdn": "srx1.ch.office.geant.net",
"site": "CCH"
},
{
"fqdn": "srx2.am.office.geant.net",
"site": "AMO"
},
{
"fqdn": "srx2.ch.office.geant.net",
"site": "CCH"
},
{
"fqdn": "rt1.kie.ua.geant.net",
"site": "KIE"
},
{
"fqdn": "rt1.fra.de.geant.net",
"site": "FRA"
},
{
"fqdn": "rt1.mar.fr.geant.net",
"site": "MAR"
},
{
"fqdn": "rt1.mil2.it.geant.net",
"site": "MIL2"
},
{
"fqdn": "rt1.por.pt.geant.net",
"site": "POR"
},
{
"fqdn": "rt1.pra.cz.geant.net",
"site": "PRA"
},
{
"fqdn": "rt0.poz.pl.geant.net",
"site": "POZ"
},
{
"fqdn": "rt2.ams.nl.geant.net",
"site": "AMS"
},
{
"fqdn": "rt0.gen.ch.geant.net",
"site": "GEN"
},
{
"fqdn": "rt2.kie.ua.geant.net",
"site": "KIE"
},
{
"fqdn": "rt2.chi.md.geant.net",
"site": "CHI"
},
{
"fqdn": "rt0.zag.hr.geant.net",
"site": "ZAG"
},
{
"fqdn": "rt2.bra.sk.geant.net",
"site": "BRA"
},
{
"fqdn": "rt0.rig.lv.geant.net",
"site": "RIG"
},
{
"fqdn": "rt0.mad.es.geant.net",
"site": "MAD"
},
{
"fqdn": "rt0.kau.lt.geant.net",
"site": "KAU"
},
{
"fqdn": "rt0.the.gr.geant.net",
"site": "THE"
},
{
"fqdn": "rt0.buc.ro.geant.net",
"site": "BUC"
},
{
"fqdn": "rt0.tar.ee.geant.net",
"site": "TAR"
},
{
"fqdn": "rt0.par.fr.geant.net",
"site": "PAR"
},
{
"fqdn": "rt0.bud.hu.geant.net",
"site": "BUD"
},
{
"fqdn": "rt0.bra.sk.geant.net",
"site": "BRA"
},
{
"fqdn": "rt0.pra.cz.geant.net",
"site": "PRA"
},
{
"fqdn": "rt0.lon2.uk.geant.net",
"site": "LON2"
},
{
"fqdn": "rt0.por.pt.geant.net",
"site": "POR"
},
{
"fqdn": "rt0.vie.at.geant.net",
"site": "VIE"
},
{
"fqdn": "rt0.cor.ie.geant.net",
"site": "COR"
},
{
"fqdn": "rt0.dub.ie.geant.net",
"site": "DUB"
},
{
"fqdn": "rt0.bru.be.geant.net",
"site": "BRU"
},
{
"fqdn": "rt0.fra.de.geant.net",
"site": "FRA"
},
{
"fqdn": "rt0.ams.nl.geant.net",
"site": "AMS"
},
{
"fqdn": "rt0.lju.si.geant.net",
"site": "LJU"
},
{
"fqdn": "rt0.sof.bg.geant.net",
"site": "SOF"
},
{
"fqdn": "rt0.ham.de.geant.net",
"site": "HAM"
}
]
[
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-7.0"
}
],
"name": "BIL-DUB-IPTRUNK",
"overlays": {
"speed": 536870912000
},
"sid": "LGS-00044",
"type": "IP TRUNK"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-1"
}
],
"name": "LONDON-LONDON-LAG-005(UNKNOWN)",
"overlays": {
"speed": 214748364800
},
"sid": "GA-000101",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-3"
}
],
"name": "LONDON-LONDON-LAG-003(UNKNOWN)",
"overlays": {
"speed": 429496729600
},
"sid": "LGA-000222",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-2"
}
],
"name": "ATHENS-LONDON-LAG-001(UNKNOWN)",
"overlays": {
"speed": 429496729600
},
"sid": "GA-000662",
"type": "ETHERNET"
},
{
"endpoints": [
{
"addresses": {
"v4": "62.40.98.60/31",
"v6": "2001:798:cc::65/126"
},
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-1.0"
}
],
"name": "LON-LON-IPTRUNK",
"overlays": {
"speed": 214748364800
},
"sid": "GS-00010",
"type": "IP TRUNK"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-3.0"
}
],
"name": "DUB-LON-IPTRUNK",
"overlays": {
"speed": 429496729600
},
"sid": "LGS-00022",
"type": "IP TRUNK"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-8"
}
],
"name": "AMSTERDAM-LONDON-LAG-001(UNKNOWN)",
"overlays": {
"speed": 429496729600
},
"sid": "LGA-00771",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-20"
}
],
"name": "JISC-AP2-LAG",
"overlays": {
"speed": 10737418240
},
"sid": "GA-01760",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-20.2501"
}
],
"name": "JISC-AP2",
"overlays": {
"speed": 10737418240
},
"sid": "GS-00480",
"type": "GEANT IP"
},
{
"endpoints": [
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "lag-20.2522"
}
],
"name": "JISC-AP2-LHCONE",
"overlays": {
"speed": 10737418240
},
"sid": "GS-02474",
"type": "L3-VPN"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-1"
}
],
"name": "UNKNOWN_NOKIA-UNKNOWN_NOKIA-LAG-001(UNKNOWN)",
"overlays": {
"speed": 214748364800
},
"sid": "LGS-000991",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-7"
}
],
"name": "UNKNOWN_NOKIA-UNKNOWN_NOKIA-LAG-002(UNKNOWN)",
"overlays": {
"speed": 429496729600
},
"sid": "LGA-000442",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20"
}
],
"name": "DUBLIN-DUBLIN-LAG-004(UNKNOWN)",
"overlays": {
"speed": 10737418240
},
"sid": "GA-50002",
"type": "ETHERNET"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20.3333"
}
],
"name": "BELNET-AP1-IAS",
"overlays": {
"speed": 10737418240
},
"sid": "GS-00524",
"type": "GEANT PEERING"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20.2433"
}
],
"name": "MARNET-AP1-IAS",
"overlays": {
"speed": 10737418240
},
"sid": "GS-00536",
"type": "GWS - INDIRECT"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20.2410"
}
],
"name": "MARNET-AP1",
"overlays": {
"speed": 10737418240
},
"sid": "GS-12489",
"type": "GEANT IP"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20:765"
},
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "pwe-34784"
}
],
"name": "$GS-50008",
"overlays": {
"speed": 10737418240
},
"sid": "GS-50008",
"type": "GEANT PLUS"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-24.0"
}
],
"name": "IX-PEERINGS-IN-INEX",
"overlays": {
"speed": 10737418240
},
"sid": "GS-00950",
"type": "IP PEERING - NON R&E (PUBLIC)"
},
{
"endpoints": [
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "lag-20:790"
},
{
"hostname": "rt0.dub.ie.geant.net",
"interface": "pwe-31757"
},
{
"hostname": "rt0.lon.uk.geant.net",
"interface": "pwe-31757"
}
],
"name": "DUBLIN-LONDON-ETHS-001(BELNET)",
"overlays": {
"speed": 10737418240
},
"sid": "GS-50035",
"type": "ETHERNET"
}
]
[
{
"city": "AMSTERDAM",
"country": "NETHERLANDS",
"latitude": 52.34638889,
"longitude": 4.93861111,
"name": "AMS"
},
{
"city": "BILBAO",
"country": "SPAIN",
"latitude": 43.3250881,
"longitude": -2.9804526,
"name": "BIL"
},
{
"city": "ATTIKI",
"country": "GREECE",
"latitude": 37.98,
"longitude": 23.73,
"name": "ATH2"
},
{
"city": "CAMBRIDGE",
"country": "UNITED KINGDOM",
"latitude": 52.19152778,
"longitude": 0.13383333,
"name": "CCH"
},
{
"city": "ATHENS",
"country": "GREECE",
"latitude": 37.97308611,
"longitude": 23.74555556,
"name": "ATH"
},
{
"city": "BRATISLAVA",
"country": "SLOVAKIA",
"latitude": 48.116833,
"longitude": 17.094389,
"name": "BRA"
},
{
"city": "BRUSSELS",
"country": "BELGIUM",
"latitude": 50.85694444,
"longitude": 4.41138889,
"name": "BRU"
},
{
"city": "BUCHAREST",
"country": "ROMANIA",
"latitude": 44.44474167,
"longitude": 26.09642222,
"name": "BUC"
},
{
"city": "DUBLIN",
"country": "IRELAND",
"latitude": 53.29198056,
"longitude": -6.41473333,
"name": "DUB"
},
{
"city": "CORK",
"country": "IRELAND",
"latitude": 51.90362538,
"longitude": -8.512314371,
"name": "COR"
},
{
"city": "BUDAPEST",
"country": "HUNGARY",
"latitude": 47.51777778,
"longitude": 19.05527778,
"name": "BUD"
},
{
"city": "CHISINAU",
"country": "MOLDOVA, REPUBLIC OF",
"latitude": 47.03072498,
"longitude": 28.82379273,
"name": "CHI"
},
{
"city": "KAUNAS",
"country": "LITHUANIA",
"latitude": 54.94067222,
"longitude": 24.01801389,
"name": "KAU"
},
{
"city": "KIEV",
"country": "UKRAINE",
"latitude": 50.44909773,
"longitude": 30.46527006,
"name": "KIE"
},
{
"city": "GENEVA",
"country": "SWITZERLAND",
"latitude": 46.2355997,
"longitude": 6.0553345,
"name": "GEN"
},
{
"city": "HAMBURG",
"country": "GERMANY",
"latitude": 53.55090278,
"longitude": 10.04648611,
"name": "HAM"
},
{
"city": "AMSTERDAM - Office",
"country": "NETHERLANDS",
"latitude": 52.31330556,
"longitude": 4.94911111,
"name": "AMO"
},
{
"city": "FRANKFURT",
"country": "GERMANY",
"latitude": 50.12029444,
"longitude": 8.73584444,
"name": "FRA"
},
{
"city": "LONDON",
"country": "UNITED KINGDOM",
"latitude": 51.4981657,
"longitude": -0.0152639,
"name": "LON"
},
{
"city": "ALCOBENDAS",
"country": "SPAIN",
"latitude": 40.53647778,
"longitude": -3.64880278,
"name": "MAD"
},
{
"city": "LISBON",
"country": "PORTUGAL",
"latitude": 38.759325,
"longitude": -9.142339,
"name": "LIS"
},
{
"city": "MARSEILLE",
"country": "FRANCE",
"latitude": 43.338019,
"longitude": 5.347666,
"name": "MAR"
},
{
"city": "LJUBLJANA",
"country": "SLOVENIA",
"latitude": 46.050171,
"longitude": 14.46033,
"name": "LJU"
},
{
"city": "SLOUGH",
"country": "UNITED KINGDOM",
"latitude": 51.5231433,
"longitude": -0.6224058,
"name": "LON2"
},
{
"city": "AUBERVILLIERS",
"country": "FRANCE",
"latitude": 48.90444444,
"longitude": 2.37111111,
"name": "PAR"
},
{
"city": "MILAN",
"country": "ITALY",
"latitude": 45.47527778,
"longitude": 9.10305556,
"name": "MIL2"
},
{
"city": "RIGA",
"country": "LATVIA",
"latitude": 56.94834444,
"longitude": 24.118,
"name": "RIG"
},
{
"city": "PORTO",
"country": "PORTUGAL",
"latitude": 41.177904,
"longitude": -8.594972,
"name": "POR"
},
{
"city": "PRAGUE",
"country": "CZECH REPUBLIC",
"latitude": 50.10166667,
"longitude": 14.39166667,
"name": "PRA"
},
{
"city": "POZNAN",
"country": "POLAND",
"latitude": 52.411775,
"longitude": 16.91756111,
"name": "POZ"
},
{
"city": "SOFIA",
"country": "BULGARIA",
"latitude": 42.67575833,
"longitude": 23.37098611,
"name": "SOF"
},
{
"city": "ZAGREB",
"country": "CROATIA",
"latitude": 45.79194444,
"longitude": 15.96944444,
"name": "ZAG"
},
{
"city": "TARTU",
"country": "ESTONIA",
"latitude": 58.383146,
"longitude": 26.71986,
"name": "TAR"
},
{
"city": "VIENNA",
"country": "AUSTRIA",
"latitude": 48.26888889,
"longitude": 16.41019444,
"name": "VIE"
},
{
"city": "BELGRADE",
"country": "SERBIA",
"latitude": 44.81141383,
"longitude": 20.39871378,
"name": "BEL"
},
{
"city": "THESSALONIKI",
"country": "GREECE",
"latitude": 40.662632,
"longitude": 22.860255,
"name": "THE"
},
{
"city": "GENEVA",
"country": "FR",
"latitude": 46.255952,
"longitude": 6.055262,
"name": "GEN2"
}
]
import json
import os
import re
import responses
from mapping_provider.api.map import RouterList, ServiceList, SiteList
DATA_DIR = os.path.join(os.path.dirname(__file__), 'data')
def _load_test_data(filename: str) -> dict:
with open(os.path.join(DATA_DIR, filename)) as f:
return json.load(f)
# @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/"
# )
@responses.activate
def test_get_sites(client):
responses.add(
method=responses.GET,
url=re.compile(r'.*/map/sites$'),
json=_load_test_data('inprov-sites.json')
)
rv = client.get("/map/sites")
assert rv.status_code == 200
assert rv.json()
SiteList.model_validate(rv.json())
@responses.activate
def test_get_routers(client):
responses.add(
method=responses.GET,
url=re.compile(r'.*/map/routers$'),
json=_load_test_data('inprov-routers.json')
)
rv = client.get("/map/routers")
assert rv.status_code == 200
assert rv.json()
RouterList.model_validate(rv.json())
@responses.activate
def test_get_trunks(client):
responses.add(
method=responses.GET,
url=re.compile(r'.*/map/services.*'),
json=_load_test_data('inprov-services.json')
)
rv = client.get("/map/trunks")
assert rv.status_code == 200
assert rv.json()
ServiceList.model_validate(rv.json())
from mapping_provider.api.common import Version
def test_version(client):
rv = client.get("/version")
assert rv.status_code == 200
assert rv.json()
Version.model_validate(rv.json())
[tox]
envlist = lint, typecheck, docs
[testenv:coverage]
description = Run unit tests and save coverage
deps =
pytest
pytest-cov
httpx # required for fastapi TestClient
responses
commands =
coverage erase
pytest --cov mapping_provider --cov-fail-under=80 --cov-report html --cov-report xml --cov-report term -p no:checkdocs
[testenv:lint]
description = Lint code with Ruff
deps = ruff
commands = ruff check mapping_provider
commands = ruff check mapping_provider test
[testenv:typecheck]
description = Type-check code with mypy
deps = mypy
deps =
mypy
types-jsonschema
types-requests
commands = mypy mapping_provider
[testenv:sbom]
description = Create SBOM for dependency analysis
deps = cyclonedx-py
commands = cyclonedx-py environment --output-format json -o bom.json
[testenv:docs]
description = Build docs
deps =
......