Function test_upcast_event_handler_arg

tests/units/test_model.py:236–260

Test that upcast event handler args work correctly. Args: handler: The handler to test. payload: The payload to test. mock_base_state_event_processor: Fixture for processing events with a BaseState. emitted_deltas: List to store emitted deltas.

(
    handler, payload, mock_base_state_event_processor, emitted_deltas
)

    ],
)
async def test_upcast_event_handler_arg(
    handler, payload, mock_base_state_event_processor, emitted_deltas
):
    """Test that upcast event handler args work correctly.

    Args:
        handler: The handler to test.
        payload: The payload to test.
        mock_base_state_event_processor: Fixture for processing events with a BaseState.
        emitted_deltas: List to store emitted deltas.
    """
    async with mock_base_state_event_processor as processor:
        await processor.enqueue(
            "test_token", Event.from_event_type(handler(**payload))[0]
        )
    assert emitted_deltas == [
        (
            "test_token",
            {
                UpcastStateWithSqlAlchemy.get_full_name(): {
                    "passed" + FIELD_MARKER: True
                }
            },
        ),
    ]


def test_no_rebind_mutable_proxy_for_instrumented_functions():