Skip to content
Snippets Groups Projects
Commit e23c01c6 authored by Erik Reid's avatar Erik Reid
Browse files

template service build

parent 1eacadd4
Branches
Tags
No related merge requests found
......@@ -33,15 +33,6 @@ class Service(BaseModel):
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]
......@@ -53,13 +44,22 @@ def _services() -> Generator[Service, None, None]:
scid_current = cache.get(inventory.REPORTING_SCID_CURRENT_CACHE_FILENAME)
poller_interfaces = cache.get(inventory.INPROV_POLLER_INTERFACES_CACHE_FILENAME)
correlator_state = cache.get(correlator.CACHED_CORRELATOR_STATE_FILENAME)
except IOError:
except FileNotFoundError:
logger.exception('not enough data available to build the service list')
return
# for service in cache.get(INPROV_MAP_SERVICES_CACHE_FILENAME):
# yield Service.from_inprov_service(service)
for service in inprov_map_services:
overlays = Overlays(speed = service['overlays']['speed'])
endpoints = list(map(Endpoint.from_inprov_endpoint, service['endpoints']))
yield Service(
sid = service['sid'],
name = service['name'],
type = service['type'],
endpoints = endpoints,
overlays = overlays,
)
def build_service_info_list() -> ServiceList:
return ServiceList(services=[])
return ServiceList(services=_services())
......@@ -7,7 +7,8 @@ import pytest
from fastapi.testclient import TestClient
from mapping_provider import create_app
from mapping_provider.backends import cache
from mapping_provider.backends import cache, inventory, correlator
from .common import load_test_data
@pytest.fixture
......@@ -44,8 +45,18 @@ def dummy_config_filename(dummy_config):
@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())
with tempfile.TemporaryDirectory() as tmp_dir:
cache.init(tmp_dir) # there's no rmq in config, so workers won't start
cache.set(inventory.INPROV_MAP_SERVICES_CACHE_FILENAME, load_test_data('inprov-services.json'))
cache.set(inventory.REPORTING_SCID_CURRENT_CACHE_FILENAME, load_test_data('scid-current.json'))
cache.set(inventory.INPROV_POLLER_INTERFACES_CACHE_FILENAME, load_test_data('poller-interfaces.json'))
cache.set(correlator.CACHED_CORRELATOR_STATE_FILENAME, load_test_data('correlator-state.json'))
with patch('sentry_sdk.init') as _mock_sentry_init:
yield TestClient(create_app())
@pytest.fixture(autouse=True)
def run_around_tests():
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment