我有一个运行循环直到数据库可用的 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
关于python - Django单元测试等待数据库,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52621819/