pyinfra/tests/payload_test.py

import pytest

from pyinfra.config import get_config
from pyinfra.payload_processing.payload import (
    QueueMessagePayload,
    get_queue_message_payload_parser,
)
from pyinfra.utils.file_extension_parsing import make_file_extension_parser


@pytest.fixture(scope="session")
def payload_config():
    return get_config()


class TestPayload:
    def test_payload_is_parsed_correctly(self, request_payload, payload_config):
        parse_payload = get_queue_message_payload_parser(payload_config)
        payload = parse_payload(request_payload)
        assert payload == QueueMessagePayload(
            dossier_id="test",
            file_id="test",
            target_file_extension="json.gz",
            response_file_extension="json.gz",
            target_file_type="json",
            target_compression_type="gz",
            response_file_type="json",
            response_compression_type="gz",
            target_file_name="test/test.json.gz",
            response_file_name="test/test.json.gz",
        )

    @pytest.mark.parametrize(
        "extension,expected",
        [
            ("json.gz", ("json", "gz")),
            ("json", ("json", None)),
            ("prefix.json.gz", ("json", "gz")),
        ],
    )
    @pytest.mark.parametrize("allowed_file_types,allowed_compression_types", [(["json", "pdf"], ["gz"])])
    def test_parse_file_extension(self, extension, expected, allowed_file_types, allowed_compression_types):
        parse = make_file_extension_parser(allowed_file_types, allowed_compression_types)
        assert parse(extension) == expected