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+unsubscr...@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/6F600904-9697-47A6-A813-9FBB6A69E7C9%40gmail.com.
For more options, visit https://groups.google.com/d/optout.
signature.asc
Description: Message signed with OpenPGP
