pyinfra/tests/conftest.py

import logging
import time
from pathlib import Path

import pytest
import testcontainers.compose

from pyinfra.config import get_config
from pyinfra.queue.queue_manager import QueueManager
from pyinfra.storage import get_storage

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

TESTS_DIR = Path(__file__).resolve().parents[0]


@pytest.fixture(scope="session", autouse=True)
def docker_compose(sleep_seconds=30):
    """Note: `autouse` can be set to `False` while working on the code to speed up the testing. In that case, run
    `docker-compose up` in the tests directory manually before running the tests.
    """
    logger.info(f"Starting docker containers with {TESTS_DIR}/docker-compose.yml...")
    compose = testcontainers.compose.DockerCompose(TESTS_DIR, compose_file_name="docker-compose.yml")
    compose.start()
    logger.info(f"Sleeping for {sleep_seconds} seconds to wait for containers to finish startup... ")
    time.sleep(sleep_seconds)
    yield compose
    compose.stop()


@pytest.fixture(scope="session")
def storage_config(client_name):
    config = get_config()
    config.storage_backend = client_name
    config.storage_azureconnectionstring = "DefaultEndpointsProtocol=https;AccountName=iqserdevelopment;AccountKey=4imAbV9PYXaztSOMpIyAClg88bAZCXuXMGJG0GA1eIBpdh2PlnFGoRBnKqLy2YZUSTmZ3wJfC7tzfHtuC6FEhQ==;EndpointSuffix=core.windows.net"
    return config


@pytest.fixture(scope="session")
def processing_config(storage_config, monitoring_enabled):
    storage_config.monitoring_enabled = monitoring_enabled
    return storage_config


@pytest.fixture(scope="session")
def bucket_name(storage_config):
    return storage_config.storage_bucket


@pytest.fixture(scope="session")
def storage(storage_config):
    logger.debug("Setup for storage")
    storage = get_storage(storage_config)
    storage.make_bucket(storage_config.storage_bucket)
    storage.clear_bucket(storage_config.storage_bucket)
    yield storage
    logger.debug("Teardown for storage")
    try:
        storage.clear_bucket(storage_config.storage_bucket)
    except:
        pass


@pytest.fixture(scope="session")
def queue_config(payload_processor_type):
    config = get_config()
    # FIXME: It looks like rabbitmq_heartbeat has to be greater than rabbitmq_connection_sleep. If this is expected, the
    #  user should not be abele to insert non working values.
    config.rabbitmq_heartbeat = config.rabbitmq_connection_sleep + 1
    return config


@pytest.fixture(scope="session")
def queue_manager(queue_config):
    queue_manager = QueueManager(queue_config)
    return queue_manager


@pytest.fixture
def request_payload():
    return {
        "dossierId": "test",
        "fileId": "test",
        "targetFileExtension": "json.gz",
        "responseFileExtension": "json.gz",
    }


@pytest.fixture(scope="session")
def response_payload():
    return {
        "dossierId": "test",
        "fileId": "test",
    }