我有一个运行循环直到数据库可用的 Django 命令:
import time
from django.db import connections
from django.db.utils import OperationalError
from django.core.management.base import BaseCommand
class Command(BaseCommand):
"""Django command to pause execution until database is available"""
def handle(self, *args, **options):
"""Handle the command"""
self.stdout.write('Waiting for database...')
db_conn = None
while not db_conn:
try:
db_conn = connections['default']
except OperationalError:
self.stdout.write('Database unavailable, waiting 1 second...')
time.sleep(0.1)
self.stdout.write(self.style.SUCCESS('Database available!'))
我想为此代码创建单元测试。
我已经成功地测试了从一开始就可用的数据库，如下所示:
def test_wait_for_db_ready(self):
"""Test waiting for db when db is available"""
with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
gi.return_value = True
call_command('wait_for_db')
self.assertTrue(True)
有没有办法测试命令在返回之前等待数据库可用？
到目前为止，我已经尝试了以下方法，但它不起作用，因为 attempt 在 getitem 之外无法访问。
def test_wait_for_db(self):
"""Test waiting for db"""
attempt = 0
def getitem(alias):
if attempt < 5:
attempt += 1
raise OperationalError()
else:
return True
with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
gi.side_effect = getitem
call_command('wait_for_db')
self.assertGreaterEqual(attempt, 5)
请您参考如下方法:
有几种方法可以实现这一点。最简单的方法可能只是放弃 getitem() 嵌套函数并使用一系列 OperationalError 设置副作用。然后，您可以使用修补后的 gi 对象的 call_count 验证尝试次数。例如:
def test_wait_for_db(self):
"""Test waiting for db"""
with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
gi.side_effect = [OperationalError] * 5 + [True]
call_command('wait_for_db')
self.assertGreaterEqual(gi.call_count, 5) # Verify using the call_count
如果您希望保留 getitem() 函数，那么我认为您只需要将 attempt 变量设为 nonlocal 即可在嵌套函数中可以看到:
def test_wait_for_db(self):
"""Test waiting for db"""
attempt = 0
def getitem(alias):
nonlocal attempt # Make the outer attempt variable visible
if attempt < 5:
attempt += 1
raise OperationalError()
else:
return True
with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
gi.side_effect = getitem
call_command('wait_for_db')
self.assertGreaterEqual(attempt, 5)
第三，正如评论中所建议的，您可以创建一个具有 attempt 属性的类，并将该类的实例用作副作用:
def test_wait_for_db(self):
"""Test waiting for db"""
class Getitem:
def __init__(self):
self.attempt = 0
def __call__(self, item):
if self.attempt < 5:
self.attempt += 1
raise OperationalError()
else:
return True
with patch('django.db.utils.ConnectionHandler.__getitem__') as gi:
getitem = Getitem()
gi.side_effect = getitem
call_command('wait_for_db')
self.assertGreaterEqual(getitem.attempt, 5) # Access the attempts from the instance
