test_integration.py 文件源码

def test_receive(pubsub_client_mock):
    messaging = queue_messaging.Messaging.create_from_dict({
        'SUBSCRIPTION': 'test-subscription',
        'MESSAGE_TYPES': [
            FancyEvent,
        ],
    })
    topic_mock = pubsub_client_mock.return_value.topic.return_value
    subscription_mock = topic_mock.subscription.return_value
    mocked_message = mock.MagicMock(
        data=(b'{"uuid_field": "cd1d3a03-7b04-4a35-97f8-ee5f3eb04c8e", '
              b'"string_field": "Just testing!"}'),
        message_id=1,
        attributes={
            'timestamp': '2016-12-10T11:15:45.123456Z',
            'type': 'FancyEvent',
        }
    )
    subscription_mock.pull.return_value = [
        (123, mocked_message)
    ]

    envelope = messaging.receive()

    topic_mock.subscription.assert_called_with('test-subscription')
    assert envelope.model == FancyEvent(
        uuid_field=uuid.UUID('cd1d3a03-7b04-4a35-97f8-ee5f3eb04c8e'),
        string_field='Just testing!'
    )

    envelope.acknowledge()

    subscription_mock.acknowledge.assert_called_with([123])