Запрос к базе данных возвращает пустоту в тестовых случаях - Django

Я обновил версию Django для своего проекта с Django-2.2.16 --> Django3.2.14.

Но с этим обновлением некоторые из моих тестовых примеров терпят неудачу, и я не могу понять причину неудачи.

Мой файл тестового примера:

import json
from os import access
from unittest import mock
from unittest.mock import patch
import asyncio
from app.models import  UserProfile
from django.test import TestCase, TransactionTestCase
from requests.models import Response

from services.check_status import check_status

loop = asyncio.get_event_loop()




@mock.patch("services.check_status.save_status")
@mock.patch("services..check_status.send_wss_message")
class TestUserStatus(TransactionTestCase):
    def setUp(self):
        super().setUp()
        self.account_id = 4
        self.sim_profile = UserProfile.objects.create()
            
    def test_check_status_completed(
        self,
        mock_send_wss_message,
        mock_save_status,
    ):
        
        mock_save_status.return_value = {}
      

        send_wss_message_future = asyncio.Future()
        send_wss_message_future.set_result(True)
        mock_send_wss_message.return_value = send_wss_message_future

    
        loop.run_until_complete(
            check_status(
                self.sim_profile.id,
            )
        )

        
        self.assertTrue(mock_save_status.called)
        self.assertTrue(mock_send_wss_message.called)
       

Мой псевдо файл check_status имеет вид :-

import logging
from app.models import  UserProfile, UserStatus
from services.constants import WebsocketGroups
from services.user.constants import USER
from app.api.serializers import UserStatusSerializer
from services.utils import send_wss_message, sim_states, log_activity


logger = logging.getLogger(__name__)


def save_status(**kwargs):
    status =  UserStatus.objects.filter(
        status_id=kwargs.get("status_id")
    ).first()
    data = kwargs
    user_status_serializer = UserStatusSerializer(status, data, partial=True)
    if user_status_serializer.is_valid():
        user_status_serializer.save()



async def check_status(
   profile_id
):
    user_profile = UserProfile.objects.get(id=profile_id)
    login_token = get_login_token(user_profile)
    user_creds = env["user_api"]

    headers = USER["headers"]
    subscription_details = SimClient.get(
        USER["url"], headers
    )
    transaction_status = subscription_details.json()["Status"]
    subscription_data = subscription_details.json()["Data"][0]
    transaction_status_details = subscription_data["TransactionStatusDetails"]
    
    error_message = ""
    status = ""
   

    if transaction_status == "Success":
        #perform some actions and save status...
        message = {
            "type": "user_profile",
            "data": [user_profile.id, transaction_status, {"results": {}},],
        }
        await send_wss_message(
            user_profile.id, message=message, group_name=WebsocketGroups.USER_PROFILE,
        )
    else: 
        #perform some actions ...

Когда я запускаю файл test-case, он создает объект UserProfile, но когда управление переходит к функции check_status в int UserProfile.objects.all возвращается <QuerySet []>.

Я сделал временную функцию синхронизации для возврата списка всех профилей пользователей и вызвал ее внутри моего test_check_status_completed, и она вернула список. Но для асинхронных функций, которые вызываются через loop.run_until_complete, все они вернули <QuerySet []>.

Вернуться на верх