Skip to content

Test 03 Message Provider

Producer test of example message.

This test will read a pact between the message handler and the message provider and then validate the pact against the provider.

Attributes

CURRENT_STATE: str | None = None module-attribute

PACT_DIR = Path(__file__).parent.parent.parent / 'pacts'.resolve() module-attribute

responses: dict[str, dict[str, str]] = {'a request to write test.txt': {'function_name': 'send_write_event'}, 'a request to read test.txt': {'function_name': 'send_read_event'}} module-attribute

Classes

Functions

message_producer_function() -> tuple[str, str]

Source code in examples/tests/v3/test_03_message_provider.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
def message_producer_function() -> tuple[str, str]:
    producer = FileSystemMessageProducer()
    producer.queue = MagicMock()

    assert CURRENT_STATE is not None, "State is not set"
    function_name = responses.get(CURRENT_STATE, {}).get("function_name")
    assert function_name is not None, "Function name could not be found"
    producer_function = getattr(producer, function_name)

    if producer_function.__name__ == "send_write_event":
        producer_function("provider_file_name.txt", "Hello, world!")
    elif producer_function.__name__ == "send_read_event":
        producer_function("provider_file_name.txt")

    return producer.queue.send.call_args[0][0], "application/json"

state_provider_function(state_name: str) -> None

Source code in examples/tests/v3/test_03_message_provider.py 
48
49
50
def state_provider_function(state_name: str) -> None:
    global CURRENT_STATE  # noqa: PLW0603
    CURRENT_STATE = state_name

test_producer() -> None

Test the message producer.

Source code in examples/tests/v3/test_03_message_provider.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def test_producer() -> None:
    """
    Test the message producer.
    """
    with start_provider(
        handler_module=__name__,
        handler_function="message_producer_function",
        state_provider_module=__name__,
        state_provider_function="state_provider_function",
    ) as provider_url:
        verifier = (
            Verifier()
            .set_state(
                provider_url / "set_provider_state",
                teardown=True,
            )
            .set_info("provider", url=f"{provider_url}/produce_message")
            .filter_consumers("v3_message_consumer")
            .add_source(PACT_DIR / "v3_message_consumer-v3_message_provider.json")
        )
        verifier.verify()