Django - Тестирование асинхронного метода приводит к тому, что соединение уже закрыто
Я хочу протестировать некоторый асинхронный метод, который я использую в асинхронном представлении, но я хочу быть предельно ясным, что сейчас я не тестирую свое представление, я уже видел документацию на django, но она только для представлений....
У меня есть класс, содержащий множество методов и даже обращающийся к базе данных асинхронно.
class SessionService:
""" Session service.
Attributes:
uuid (str): system uuid
gcloud (str): console gcloud_id
start (str): start time of the period
end (str): end time of the period
"""
def __init__(self, uuid: str, gcloud: str, start: str, end: str):
self.system_uuid: str = uuid
self.gcloud_id: str = gcloud
self.start: str = start
self.end: str = end
@database_sync_to_async
def __vision_data(self) -> Tuple[List[str], List[str], List[int]]:
""" Async method to get vision module's data from a uuid
Returns
----------
QuerySet
List where 1st element are gcloud id, 2nd UUID and last system IDs
"""
return System.objects.get_vision_conf_data(self.system_uuid)
@async_property
async def __vision_system_ids(self) -> List[int]:
""" get visions Ids
Returns:
List[int]
"""
data = await self.__vision_data()
return data[2]
async def get_session_list(self) -> List[Dict[str, Union[str, bool, int]]]:
""" Get all session for a period for a system
Returns:
List[Dict[str, Union[str, bool, int]]]
"""
spraying_session: SprayingSessionSumUpGetter = SprayingSessionSumUpGetter(self.gcloud_id, self.start,
self.end)
vision_systems_ids: List[int] = await self.__vision_system_ids
sessions: List[Dict[str, Union[str, pd.Timestamp]]] = await spraying_session.get_sum_up(
vision_systems_ids)
return sessions
У меня есть две разные ситуации с одной и той же проблемой в конце.
Ситуация ОДНА:
Сейчас я пытаюсь протестировать get_session_list. Для этого я использую TestCase из django.test .
Мои тесты наследуются от другого класса, потому что мне нужно установить экземпляр System, и это зависит от множества других таблиц, поэтому я управляю этим на другом тесте Класс SystemSetUpTest Итак, у меня есть:
class SystemSetUpTest(TestCase):
@staticmethod
def setUp(): # pylint: disable=too-many-locals
create_everything_needed
@staticmethod
def tearDown():
delete_all_stuff
class SystemServicesTest(SystemSetUpTest):
def setUp(self):
super().setUp()
вот тест, вызывающий проблему:
async def test_getting_sessions_list(self):
uuid: str = self.uuid
console_gcloud_id = 'console'
start_period = '2021-05-06'
end_period = '2021-05-07'
service = SessionService(uuid, console_gcloud_id, start_period, end_period)
print(await service.get_session_list())
и вот полная трассировка:
Traceback (most recent call last):
File "/app/systems/tests/test_set_up.py", line 60, in tearDown
User.objects.filter(email="jul@gmail.com").delete()
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 745, in delete
collector.collect(del_query)
File "/usr/local/lib/python3.8/dist-packages/django/db/models/deletion.py", line 243, in collect
new_objs = self.add(objs, source, nullable,
File "/usr/local/lib/python3.8/dist-packages/django/db/models/deletion.py", line 107, in add
if not objs:
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 284, in __bool__
self._fetch_all()
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 1324, in _fetch_all
self._result_cache = list(self._iterable_class(self))
File "/usr/local/lib/python3.8/dist-packages/django/db/models/query.py", line 51, in __iter__
results = compiler.execute_sql(chunked_fetch=self.chunked_fetch, chunk_size=self.chunk_size)
File "/usr/local/lib/python3.8/dist-packages/django/db/models/sql/compiler.py", line 1167, in execute_sql
cursor = self.connection.cursor()
File "/usr/local/lib/python3.8/dist-packages/django/utils/asyncio.py", line 26, in inner
return func(*args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 259, in cursor
return self._cursor()
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 237, in _cursor
return self._prepare_cursor(self.create_cursor(name))
File "/usr/local/lib/python3.8/dist-packages/django/db/utils.py", line 90, in __exit__
raise dj_exc_value.with_traceback(traceback) from exc_value
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/base/base.py", line 237, in _cursor
return self._prepare_cursor(self.create_cursor(name))
File "/usr/local/lib/python3.8/dist-packages/django/utils/asyncio.py", line 26, in inner
return func(*args, **kwargs)
File "/usr/local/lib/python3.8/dist-packages/django/db/backends/postgresql/base.py", line 236, in create_cursor
cursor = self.connection.cursor()
django.db.utils.InterfaceError: connection already closed
В принципе, он может снести базу данных, мы можем увидеть это из:
File "/app/systems/tests/test_set_up.py", line 60, in tearDown User.objects.filter(email="jul@gmail.com").delete()
ситуация ДВА класс SystemServicesTest наследуется непосредственно от TestCase и я устанавливаю свою систему непосредственно в его методе setUp,
тогда трассировка стека будет такой:
Проблема, похоже, возникает из-за:
File "/app/systems/models/manager/manager.py", line 130, in get_vision_conf_data gcloud: List[str] = [data[0] for data in visions]
>
Вот полный метод:
def get_vision_conf_data(self, system_uuid: str) -> Tuple[List[str], List[str], List[int]]:
""" Get vision conf data for a system
Parameters
----------
system_uuid : str
The system uuid
Returns
-------
Tuple[List[str], List[str], List[int]]
List[str]: List of all visions gcloud id
List[str]: List of all visions uuid
List[str]: List of all visions system_id
"""
visions: QuerySet = self.get_queryset().get_vision_conf_details(system_uuid)
gcloud: List[str] = [data[0] for data in visions]
uuid: List[str] = [data[1] for data in visions]
system_id: List[int] = [data[2] for data in visions]
return gcloud, uuid, system_id
где get_vision_conf_details - это вызов моей базы данных.
Как мне протестировать мой класс? Любая помощь будет замечательной! Спасибо.