如何在 Django 中测试关闭数据库连接的方法？

Question

我有一个很长的运行 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 会导致这些测试失败，因为 django 的 TestCase class 将测试包装在一个事务，并关闭该事务内的连接会导致事务中断并引发 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方法的最佳方法是什么？

Answer 1

好吧，我不确定这是否是最佳答案，但由于到目前为止这个问题还没有得到回应，我将 post 我最终用于 posterity 的解决方案:

我创建了一个辅助函数，用于在关闭连接之前检查我们当前是否处于原子块中：

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 在测试中被调用，因此我们不再中断测试事务。

Answer 2

Django 有一个 TransactionTestCase class，它不会将测试代码包含在数据库事务中，这与 TestCase 不同，后者是。希望这有帮助。

Answer 3

您可以从 TransactionTestCase 继承您的测试 class 而不是修改 myprocess 函数，这首先不会将您的测试函数包装在事务原子块中。如果你热衷于在你的测试函数中使用原子事务，那么只在你需要的地方包装它。

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

ps：从 TestCase 继承您的测试 class 将使用事务原子块包装您的总测试函数。

如何在 Django 中测试关闭数据库连接的方法？

How do I test a method in Django which closes the database connection?

django

django-testing

django-database