Как протестировать метод в Django, который закрывает соединение с базой данных?

У меня есть длительный процесс Python, который использует Django ORM. Это выглядит примерно так:

import django
from django.db import connection

from my_app import models


def my_process():
    django.setup()
    while (True):
        do_stuff()
        connection.close()
        models.MyDjangoModel().save()

Иногда do_stuff занимает очень много времени, и в этот момент я столкнулся с ошибкой, когда время ожидания моего соединения MySql истекло, потому что сервер базы данных прервал соединение как бездействующее. Добавление строки connection.close() заставляет django каждый раз получать новое соединение и устраняет эту проблему. (См. https://code.djangoproject.com/ticket/21597).

Однако я тестирую этот процесс, используя django.test.TestCase, и вызов connection.close приводит к сбою этих тестов, поскольку класс TestCase django оборачивает тест в транзакцию, а закрытие соединения в этой транзакции приводит к разрыву транзакции и вызывает django.db.transaction.TransactionManagementError.

Одна из попыток решить эту проблему, которую я пробовал, заключается в установке параметра базы данных CONN_MAX_AGE и вызове вместо этого connection.close_if_unusable_or_obsolete, но транзакция также изменяет параметр autocommit соединения со значения по умолчанию True на False, что, в свою очередь, заставляет close_if_unusable_or_obsolete попытаться закрыть соединение в любом случае. (https://github.com/django/django/blob/master/django/db/backends/base/base.py#L497).

Я думаю, я мог бы также издеваться над connection.close в тесте, чтобы он ничего не делал, но это выглядит как хакерство.

Каков наилучший способ протестировать метод django, который должен закрыть соединение с базой данных?


django django-testing django-database
person Rahul Gupta-Iwasaki    schedule 30.08.2016    source источник

Ответы (3)


arrow_upward
9
arrow_downward

Что ж, я не уверен, что это лучший ответ, но, поскольку на этот вопрос до сих пор не было ответов, я опубликую решение, которое я использовал для потомков:

Я создал вспомогательную функцию, которая проверяет, находимся ли мы в данный момент в атомарном блоке перед закрытием соединения:

import django
from django.db import connection

from my_app import models


def close_connection():
    """Closes the connection if we are not in an atomic block.

    The connection should never be closed if we are in an atomic block, as
    happens when running tests as part of a django TestCase. Otherwise, closing
    the connection is important to avoid a connection time out after long actions.     
    Django does not automatically refresh a connection which has been closed 
    due to idleness (this normally happens in the request start/finish part 
    of a webapp's lifecycle, which this process does not have), so we must
    do it ourselves if the connection goes idle due to stuff taking a really 
    long time.
    """
    if not connection.in_atomic_block:
        connection.close()


def my_process():
    django.setup()
    while (True):
        do_stuff()
        close_connection()
        models.MyDjangoModel().save()

Как говорится в комментарии, close_connection предотвращает вызов connection.close при тестировании, поэтому мы больше не прерываем тестовую транзакцию.

person Rahul Gupta-Iwasaki    schedule 05.09.2016

arrow_upward
1
arrow_downward

Вместо изменения функции myprocess вы можете наследовать свой тестовый класс от TransactionTestCase который в первую очередь не оборачивает вашу тестовую функцию в атомарный блок транзакции. Если вы заинтересованы в том, чтобы транзакция была атомарной в вашей тестовой функции, оберните ее только там, где вам это нужно.

class Test(TransactionTestCase):
     def test_function(self):
         do_stuff()
         with transaction.atomic():
             your_transaction_atomic_test()
         my_process()
         do_other_stuff()

ps: наследование вашего тестового класса от TestCase обернет вашу общую тестовую функцию атомарным блоком транзакции.

person Sandeep    schedule 20.02.2020

arrow_upward
0
arrow_downward

В Django есть класс TransactionTestCase, который НЕ включает в себя тестовый код в транзакции базы данных, в отличие от TestCase, который делает это. Надеюсь это поможет.

person Sergei Krupenin    schedule 28.02.2019