conftest.py

import json
import os
import tempfile
from unittest.mock import patch

import pytest
from fastapi.testclient import TestClient

from mapping_provider import create_app
from mapping_provider.backends import cache, inventory, correlator
from .common import load_test_data


@pytest.fixture
def dummy_config():
    return {
        'sentry': {
            'dsn': 'https://token@hostname.geant.org:1111/a/b',
            'environment': 'unit tests'
        },
        'inventory': 'https://inventory.bogus.domain',
        'reporting': 'http://reporting.another-bogus.domain',
        # no rmq param to skip correlator thread
        # 'rmq': {
        #     'brokers': [
        #         'test-noc-alarms01.geant.org',
        #         'test-noc-alarms02.geant.org',
        #         'test-noc-alarms03.geant.org'
        #     ],
        #     'username': 'guest',
        #     'password': 'guest',
        #     'vhost': '/'
        # }
    }


@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 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():
    assert cache._cache_dir is None  # test env sanity check
    yield

    # make sure cache is set to unused before the next test
    cache._cache_dir = None