Failing test when calling django orm code wrapped in database_sync_to_async

[tomek-rej]

I'm trying to test my Channels consumer which calls database_sync_to_async code.
The consumer looks something like this:

class MyConsumer(AsyncJsonWebsocketConsumer):
    async def connect(self):
        my_obj = await self.get_obj()
        ...other code

    @database_sync_to_async
    def get_obj(self):
        return MyModel.objects.get(filter_condition)

The test is using the @pytest.mark.asyncio and @pytest.mark.django_db decorators ie:

@pytest.mark.asyncio
@pytest.mark.django_db
async def test_heartbeat():
    communicator = WebsocketCommunicator(MyConsumer, '<path>')
    await communicator.connect()
    await communicator.disconnect()

I'm using the following command to run the test:

./manage.py test xxx/tests.py::test_heartbeat

The test itself passes, however at the end of the test run I always get the following error:

=============================================== ERRORS ===============================================
________________________________ ERROR at teardown of test_heartbeat _________________________________

self = <django.db.backends.utils.CursorWrapper object at 0x7fbc7b8d80b8>, sql = 'DROP DATABASE "test"'
params = None
ignored_wrapper_args = (False, {'connection': <django.db.backends.postgresql.base.DatabaseWrapper object at 0x7fbc7b0481d0>, 'cursor': <django.db.backends.utils.CursorWrapper object at 0x7fbc7b8d80b8>})

    def _execute(self, sql, params, *ignored_wrapper_args):
        self.db.validate_no_broken_transaction()
        with self.db.wrap_database_errors:
            if params is None:
>               return self.cursor.execute(sql)
E               psycopg2.OperationalError: database "test" is being accessed by other users
E               DETAIL:  There is 1 other session using the database.

I can make the test failure go away by removing all references to database_sync_to_async in the consumer, but my understanding is that is poor practice to have sync code (like calling django orm) running inside an async function.

Strangely when I get the failing test two tests run (one pass and one fail), but when I remove the references to database_sync_to_async only one test runs.

Here are the versions of my libraries:

django==2.0.6
daphne==2.2.0
asgiref==2.3.2
channels-redis==2.2.1
pytest==3.6.1
pytest-asyncio==0.8.0
pytest-django==3.1.2

[andrewgodwin]

I'm guessing that the pytest.mark.django_db decorator from pytest-django is probably interacting with Channels here. What happens if you remove that?

[tomek-rej]

I removed the pytest.mark.django_db decorator and removed all calls to the django orm and the test passed.
Then I brought back the decorator, but didn't reinstate any calls to the orm. The test still passed.
Finally I brought back the call to the orm, and the test failed.

So the tests can pass if the decorator is there, but only if no django orm calls are made.

Also if I remove the pytest.mark.django_db and try to call the orm I get the error:

E       Failed: Database access not allowed, use the "django_db" mark, or the "db" or "transactional_db" fixtures to enable it.

so I just assumed decorator needs to be there if I want my tests to make any orm calls.

[andrewgodwin]

Hm. Then yes, this is a bad interaction between pytest-django and Channels, as they both do thing to try and make the ORM function properly outside of a request. Try just the db mark, that might do less?

[tomek-rej]

If I used the pytest.mark.db instead of pytest.mark.django_db I got this error:

E       Failed: Database access not allowed, use the "django_db" mark, or the "db" or "transactional_db" fixtures to enable it.

So instead I tried to use the db fixture as the alternative as suggested in the error.
Here is some minimal code which still fails:

from channels.db import database_sync_to_async
import pytest
from accounts.models import User

@database_sync_to_async
def get_user():
    return User.objects.first()

@pytest.mark.asyncio
@pytest.fixture()
async def test_x(db):
    await get_user()

I get the same error as before:

E               django.db.utils.OperationalError: database "test" is being accessed by other users
E               DETAIL:  There is 1 other session using the database.

Now if I remove the call to @database_sync_to_async the test passes:

from channels.db import database_sync_to_async
import pytest
from accounts.models import User

@pytest.mark.asyncio
@pytest.fixture()
async def test_x(db):
    user = User.objects.first()

[tomek-rej]

The good news is I think I found a way to have the test pass.
Until now I was just running that individual test case, but I noticed that if I run the whole test suite (which had other tests other than this one) the test passes.
So I think this issue only arises if this type of test is the last one to run (or is the only one to run).

[andrewgodwin]

Hm, this is going to need some more investigation by someone to figure out if we can make the two cooperate. For now I'd recommend trying to avoid these kinds of tests or figuring out a workaround.

[vijayshan]

@tomek-rej
Have you tried setting the transaction parameter on the pytest.mark.django_db decorator to true?

This is a sample test:
`@pytest.mark.django_db(transaction=True)
@pytest.mark.asyncio
async def test_my_consumer_normal():

user = User.objects.get(username="vijayshan")
assert user is not None
user1 = User.objects.get(username="shan")
assert user1 is not None



communicator = create_communicator_for_url_and_user("/ws/chat/room1/vijayshan/", user)
communicator1 = create_communicator_for_url_and_user( "/ws/chat/room1/vijayshan1/", user1)
connected, subprotocol = await communicator.connect()
connected1, subprotocol = await communicator1.connect()
assert connected is True
assert connected1 is True
room = Room.objects.get(name="room1")
assert room is not None
# Test sending text
message_sent = "hello"
await communicator.send_to(text_data=json.dumps({"message": message_sent}))
await confirm_response_for_communicator(communicator, "vijayshan: hello")
await confirm_response_for_communicator(communicator1,"vijayshan: hello")

msg = Message.objects.get(message=message_sent)
assert msg.special_message == False
assert msg.message == message_sent
assert msg.room_id == room.id
assert msg.user_id == user.id

await communicator.disconnect()
await communicator1.disconnect()`

and this is the consumer:
`class ChatConsumer(AsyncWebsocketConsumer):
"""
ChatConsumer: The chat consumer class is responsible for all chat interactions
"""

async def connect(self):
    """
    the method that handles the initial connections. the method captures the initial room name
    and generates a room group name. if the scope has a user name then the connection is accepted
    else it is rejected. a unique id for the user is also generated.
    :return:
    """
    self.requires_history = True
    self.room_name = self.scope['url_route']['kwargs']['room_name']

    self.room_group_name = 'chat_%s' % self.room_name

    if("user" in self.scope.keys() and self.scope['user'] is not None):
        self.user_name = self.scope['user'].username

        room, created = await self.retrieve_room_name(self.room_name)
        self.room = room
        self.user = self.scope['user']
        self._id = str(uuid.uuid4())
        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()
    else:
        await self.close()

@database_sync_to_async
def retrieve_room_name(self, room_name):
    """
    This method retrieves the room name name from the db if it exists else creates one
    :param room_name: the room name that has been requested
    :return: the room name from the db and if it has been created or did it already exist
    """
    room, created = Room.objects.get_or_create(name=room_name)
    return room, created

def add_message(self, message, is_special=False):
    """
    adds a message to the database
    :param message: the message text to be added
    :param is_special: is this is a special message or not
    :return: return the message object that was just added
    """
    mess = Message(message=message, room=self.room, user=self.user, special_message=is_special)
    mess.save()
    return mess

async def disconnect(self, close_code):
    """
    Leave the room
    :param close_code:
    :return:
    """
    await self.channel_layer.group_discard(
        self.room_group_name,
        self.channel_name
    )

# Receive message from WebSocket

async def receive(self, text_data=None, bytes_data=None):
    text_data_json = json.loads(text_data)
    message = text_data_json["message"]
    if "special_message" in message:
        # Send message to room group

        await self.handle_special_message(message)
    else:
        await self.handle_regular_message(message)

async def handle_regular_message(self, message):
    message_obj = await database_sync_to_async(self.add_message)(
        message=message
    )
    message = "{}: {}".format(self.user_name, message_obj.message)
    # Send message to room group
    await self.channel_layer.group_send(
        self.room_group_name,
        {
            'type': 'chat_message',
            'message': message,
            'is_special': False
        }
    )

async def handle_special_message(self, message):
    message_obj = await database_sync_to_async(self.add_message)(
        message=message.replace("special_message ", ""),
        is_special=True
    )
    message = "{}: {}".format(self.user_name, message_obj.message)
    await self.channel_layer.group_send(
        self.room_group_name,
        {
            'type': 'special_message',
            'message': message,
            'id': self._id,
            'is_special': True
        }
    )

# Receive message from room group
async def chat_message(self, event):
    message = event['message']
    # Send message to WebSocket
    await self.send(text_data=json.dumps({
        'type': 'chat_message',
        'message': message,
    }))

async def special_message(self,event):
    message = event['message']
    id = event['id']
    # Send message to socket
    if self._id != id:
        await self.send(text_data=json.dumps({
            'type': 'chat_message',
            'message': message,
        }))

`
My tests are passing.

[tomek-rej]

@vijayshan thanks for your feedback.
I did try to use transaction=True in my pytest.mark.django_db, but didn't seem to help.
However I did not try to run your example code exactly, so it's possible there's something specific to my database that's causing those errors.

One thing I noticed was with just a single test case, it passes but as soon as you add more test cases there's a bigger chance some connections won't be closed properly resulting in the error message from my original post.
Also in case it helps there's another issue someone created in the pytest-asyncio github that exactly describes my problem: pytest-dev/pytest-asyncio#82.
Unfortunately no solution there either.

Anyway in my case I came up with a really hacky fix.
It involves calling raw sql to close those connections just before the last test ends. I'm using postgres so my code looks like:

def close_db_connections():
    sql = "select * from pg_stat_activity where datname = 'test'"
    with connection.cursor() as cursor:
        cursor.execute(sql)
        for row in cursor.fetchall()[0:-1]:
            cursor.execute(
                'select pg_terminate_backend({}) from pg_stat_activity'.format(
                    row[2]))

Kind of ugly, I know, but at least now my tests are passing all the time.

[adamhooper]

Here's my ugly hack:

    def setUp(self):
        super().setUp()

        # We'll execute with a 1-worker thread pool. That's because Django
        # database methods will spin up new connections and never close them.
        # (@database_sync_to_async -- which execute uses --only closes _old_
        # connections, not valid ones.)
        #
        # This hack is just for unit tests: we need to close all connections
        # before the test ends, so we can delete the entire database when tests
        # finish. We'll schedule the "close-connection" operation on the same
        # thread as @database_sync_to_async's blocking code ran on. That way,
        # it'll close the connection @database_sync_to_async was using.
        self._old_loop = asyncio.get_event_loop()
        self.loop = asyncio.new_event_loop()
        self.loop.set_default_executor(ThreadPoolExecutor(1))
        asyncio.set_event_loop(self.loop)

    # Be careful, in these tests, not to run database queries in async blocks.
    def tearDown(self):
        def close_thread_connection():
            # Close the connection that was created by @database_sync_to_async.
            # Assumes we're running in the same thread that ran the database
            # stuff.
            django.db.connections.close_all()

        self.loop.run_in_executor(None, close_thread_connection)

        asyncio.set_event_loop(self._old_loop)

        super().tearDown()

... and then instead of async_to_async(func)(params), I run self.loop.run_until_complete(func(params)).

This sort of test case should be the exception, not the norm. You can usually structure your code so that each single test either tests scheduling or database queries but not both.

[jpic]

Is there a related issue on the pytest-django repo ?