データベース接続を閉じるDjangoのメソッドをテストするにはどうすればよいですか？

Question

私はDjango ORMを使用するPythonの長いプロセスを持っています。時にはdo_stuffは、データベース・サーバがアイドルとして接続を殺したので、わたしのMySQL接続がタイムアウトエラーにその時点で私が実行していた、本当に長い時間がかかる

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() 

：それは次のようになります。 connection.close()行を追加すると、djangoは毎回新しい接続を取得し、その問題を修正します。 （https://code.djangoproject.com/ticket/21597参照）。

しかし、私はdjango.test.TestCaseを使用してこのプロセスをテストしていて、DjangoのTestCaseクラスは、トランザクション内でテストを包むようconnection.closeを呼び出すと、これらのテストが失敗し、そのトランザクション内の接続を閉じると、トランザクションが破損する原因となり、A提起django.db.transaction.TransactionManagementError。

私はCONN_MAX_AGEデータベースパラメータを設定し、代わりにconnection.close_if_unusable_or_obsoleteを呼び出しますが、トランザクションはまた今度にしようとするclose_if_unusable_or_obsoleteを引き起こしFalseからTrueの設定のデフォルト値からの接続のautocommit設定を変更されて試してみました。この問題を解決する一つの試みととにかく接続を閉じてください（https://github.com/django/django/blob/master/django/db/backends/base/base.py#L497）。

私もconnection.closeを試してみることができますので、何もしませんが、それはハッキーのようです。

データベース接続を閉じる必要があるdjangoメソッドをテストする最良の方法は何ですか？

Answer 1

まあ、私はこれが最善の答えがあるかどうかわからないんだけど、この質問は、これまでのところ応答を確認していませんから、私は後世のために使用して終了ソリューション掲載します：私は、ヘルパー関数を作成し

をこれは私たちが接続を閉じる前に、アトミック・ブロックに現在あるかどうかをチェックします。コメントの状態として

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を防止していないので、私たちは、もはやテストトランザクションを破ります。