Re: [django-channels] Testing events from post_save signal

Here's the gist (https://gist.github.com/tricoder42/af3d0337c1b33d82c1b32d12bd0265ec) with consumer.

Dne neděle 25. února 2018 15:37:19 UTC+1 Tomáš Ehrlich napsal(a):

Hello,

I've just migrated my project to django-channels 2.x. Thanks to everyone involved!

I'm trying to write a test for a consumer.

I have a post_save signal receiver, which sends a message to a group. As I understand,

I need to wrap `group_send` with `async_to_sync` because django signals can't be

async functions:

def notify_on_model_changes(model):
    from django.contrib.contenttypes.models import ContentType
    ct = ContentType.objects.get_for_model(model)
    model_label = '.'.join([ct.app_label, ct.model])

    channel_layer = get_channel_layer()

    group_send = async_to_sync(channel_layer.group_send)

    def receiver(sender, instance, **kwargs):
        payload = {
            'type': 'model.changed',
            'pk': instance.pk,
            'model': model_label
        }     
        group_send(f'django.{model_label}', payload)

    post_save.connect(receiver, sender=model, weak=False,
                      dispatch_uid=f'django.{model_label}')

# in AppConfig.ready:

# notify_on_model_changes(Conversation)

My test suite, however, is async function:

@pytest.fixture
async def communicator():
    communicator = WebsocketCommunicator(GraphqlSubcriptionConsumer, "/")
    await communicator.connect()
    yield communicator
    await communicator.disconnect()
    
async def test_subscription_start(communicator):
    def make_conversation():
        return Conversation.objects.create()

    # function body truncated

    # subscribe for changes in Conversation model
    await communicator.send_json_to(data)

    conversation = await sync_to_async(make_conversation)()
    response = await communicator.receive_json_from()
    assert response['type'] == 'data'

I can't use `Conversation.objects.create()` directly, because it uses `async_to_sync`.

First, I need to convert it to async and await the result. I kinda feel this is hackish

jumping from async to sync and back to async, but so far everything works as expected

and test works.

Here comes the punchline:

The tests fail to teardown cleanly, because apparently there's hanging DB connection

and after a while pytest just fails with `There is 1 other session using the database.`.

Breadcrumbs:

1. If I comment out last three lines of test (make_conversations and waiting for result),

the test exits cleanly - seems like there's no problem with passing `sync_to_async` function

to `post_save.connect`.

2. If I create `async_make = sync_to_async(make_conversation)`, but don't call it at all,

the test exists cleanly - I thought that there might be problem with calling `async_to_sync`

inside code wrapped with `sync_to_async`.

I suspect there's a hanging db connection which isn't cleaned and/or garbage collected.

I would also appreciate any comments about structure of such tests - is there cleaner way

test django signals inside async test cases?

Cheers,

   Tom

--
You received this message because you are subscribed to the Google Groups "Django users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to django-users+unsubscribe@googlegroups.com.
To post to this group, send email to django-users@googlegroups.com.
Visit this group at https://groups.google.com/group/django-users.
To view this discussion on the web visit https://groups.google.com/d/msgid/django-users/af745c4f-7571-40d3-b738-1593d4d75bbb%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.