Django:如何在使用cron作业时测试并发问题
Django:如何在使用cron作业时测试并发问题
提问于 2021-11-03 11:14:06
我有一个Django应用程序托管在远程服务器上，它以相对较短的间隔运行一些cron作业。其中一个cron作业执行一个命令，从数据库中获取一个查询集，调用一个外部API，并根据API的响应更改模型。如果我不小心，cron作业将在API响应之前执行多次，从而导致并发问题和同一模型的多个实例同时更新。
为了避免这个问题，我有不同的策略，但我想编写可以在本地运行的测试，模拟API调用，并确保两个cron任务不会同时处理一个对象。我该怎么做？
我的代码如下所示(展示问题的说明性目的)：
def task():
qs = MyModel.objects.filter(task_ran=False)
for model in qs:
resp = api_call(model.foo)
model.bar = resp
model.task_ran = True
model.save()那么，我如何编写一个测试，以检查如果task()在第一个调用完成之前第二次被调用，那么它就不会再次更新模型，API也不会再次被调用？下面是一个测试的草图，我尝试将对task()的调用放在单独的线程中，但是这会导致测试冻结--在KeyboardInterrupt之后--失败
django.db.utils.OperationalError:其他用户正在访问数据库"test_db“
详细信息:还有一个使用数据库的其他会话。`
@patch("api_call")
def test_task(self, mock_api_call):
def side_effect(number):
time.sleep(2)
return number + 1
mock_api_call.side_effect = side_effect
# how to call these simultaneously? threading causes Django to get mad
task()
task()
mock_api_call.assert_called_once()回答 1
Stack Overflow用户
回答已采纳
发布于 2021-11-04 13:10:27
好吧，我找到了一个基于this answer的答案。基本上，可以通过线程在Django中进行测试，但是它需要几件事情：
首先，测试类本身必须是does).
- Secondly，的子类(至少如果它涉及到任何数据库恶作剧或使用代码的
.select_for_update，那么在每个线程中打开的数据库连接应该在线程终止时再次关闭。这可以通过使用一个.add_done_callback.来创建Future，然后添加一个回调函数来完成，当线程通过Future完成时，就可以这样做。
因此，使用此方法，可以编写如下测试：
import concurrent.futures
from django.db import connections
from django.test import TransactionTestCase
class CronTestCase(TransactionTestCase):
def on_done(self, future):
connections.close_all()
@patch("api_call")
def test_task(self, mock_api_call):
# setup the test
num_threads = 5
with concurrent.futures.ThreadPoolExecutor() as executor:
for _ in range(num_threads):
future = executor.submit(task)
future.add_done_callback(self.on_done)
mock_api_call.assert_called_once()页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/69823865
复制相关文章
相似问题
腾讯云开发者
