- html - 出于某种原因,IE8 对我的 Sass 文件中继承的 html5 CSS 不友好?
- JMeter 在响应断言中使用 span 标签的问题
- html - 在 :hover and :active? 上具有不同效果的 CSS 动画
- html - 相对于居中的 html 内容固定的 CSS 重复背景?
我创建了一个依赖于其他模块的 celery 任务。我在模型保存方法中延迟了 celery 任务,我想在没有 celery 的情况下通过所有测试用例。下面是我的代码。
def save(self, force_insert=False, force_update=False, using=None,
update_fields=None):
super(Offers, self).save(force_insert=False, force_update=False, using=None,
update_fields=None)
change_offer_status.apply_async(args=[self.id], eta=self.valid_to, queue='credr_core_task',
routing_key='credr_core_task')
测试.py
class OfferTests(APITestCase):
authenticated_client = APIClient()
def setUp(self):
data = {
"username": "vivek",
"first_name": "Vivek",
"last_name": "Dogra",
"email": "vivk@credr.com",
"contact_number": "9834982602",
"password": "easy"
}
self.user = User.objects.create(data)
token = Token.objects.get(user__username='vivek')
self.authenticated_client.credentials(HTTP_AUTHORIZATION='Token ' + token.key)
mock_task = Mock()
mock_task.get = Mock(return_value={'success': True})
print mock_task.get() # outputs {'success': True}
with patch('offers.tasks.change_offer_status.apply_async', new=mock_task) as mocked_task:
mocked_task.return_value = True
@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
CELERY_ALWAYS_EAGER=True, )
def test_add_offer(self):
"""
add user address
"""
start_date = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
end_date = datetime.strftime(datetime.now() + timedelta(days=3), "%Y-%m-%d %H:%M:%S")
data = {
"type": OfferTypeEnum.FLAT._value_,
"name": "CredR Offer",
"code": "FLAT100",
"status": OfferStatusEnum.ACTIVE._value_,
"value": "1004",
"discount_value": 1004,
"valid_from": start_date,
"valid_to": end_date,
"message": "GET FLAT 100"
}
response = self.authenticated_client.post(URL, data, format='json')
self.assertEqual(response.status_code, status.HTTP_201_CREATED)
我在setUp方法中创建了一个模拟对象。但我收到rabbitMq的连接拒绝错误。
最佳答案
i want to pass all test cases without celery
运行测试时设置CELERY_ALWAYS_EAGER = True
。
[...] tasks will be executed locally instead of being sent to the queue.
在这里阅读其他配置选项也可能是值得的:http://docs.celeryproject.org/en/latest/configuration.html#celery-always-eager
关于python - 模拟 Celery 任务方法 apply_async,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/35284043/
我是 python 的新手,我有一个函数可以为我的数据计算特征,然后返回一个应该处理并写入文件的列表。,..我正在使用 Pool 进行计算,然后使用写入文件的回调函数,但是回调函数没有被调用,我已经在
我试图了解多进程池是如何工作的。在下面的编程中,我创建了一个包含 4 个进程的池。 我使用回调函数调用 apply_async ,该函数应该更新名为 result_list 的列表 import Qu
我希望如果我调用 apply_async在实例方法中并获得其结果,所做的任何更改都将保留为 fork 进程的一部分。但是,似乎每次对 apply_async 的新调用都会创建所述实例的新副本。 采取以
我只是在使用 Python 的多处理模块,但是以下代码正在连续运行,但没有给出任何输出。我究竟做错了什么?我也尝试过 pool.close() 和 pool.join() 但没有效果。 这是我迄今为止
我尝试通过 apply_async 将共享计数器传递给多处理中的任务,但它失败并出现以下错误“RuntimeError:同步对象只能通过继承在进程之间共享”。这是怎么回事 def processLin
我有一个带有 apply_async 的进程池,其中不同的进程需要不同的时间来提供输出。一旦一个进程完成,我就会对其输出进行一些计算。在我想启动另一个进程之后。通过这种方式,我想创建一个无限循环,它启
这是我的代码: import multiprocessing import time import os def worker(): print str(os.getpid()) + " is
使用覆盖率来查看必须测试的内容,并且覆盖率显示旁边必须要测试的内容:send_alert.apply_async() 我知道是celery任务,但是有什么办法可以测试这行代码吗? 理解测试逻辑的其余代
传递给多处理的 apply_async() 的函数内的 print() 不会打印任何内容。 我想最终使用 apply_async 来处理大块的文本文件。因此,我希望脚本在屏幕上打印出已经处理了多少行。
我正在计算大量函数(大约1000000),并且由于它非常耗时,所以我使用multiprocessing.Pool.apply_async函数。但是,当我尝试使用 AsyncResult 类的 .get
import time import multiprocessing def multi_thread(files): q = multiprocessing.Queue() for
我一直在关注文档以尝试了解多处理池。我想到了这个: import time from multiprocessing import Pool def f(a): print 'f(' + st
我在我的应用程序中使用 celery 来运行周期性任务。让我们看下面的简单示例 from myqueue import Queue @perodic_task(run_every=timedelta(
我正在尝试了解使用多处理池的 apply_sync 方法时幕后发生的事情。 谁运行回调方法?是调用apply_async的主进程吗? 假设我发送了一大堆带有回调的 apply_async 命令,然后继
我有一个脚本,其中包括从列表中打开一个文件,然后对该文件中的文本执行某些操作。我正在使用 python 多处理和 Pool 来尝试并行化此操作。脚本的抽象如下: import os from mult
如果我理解正确,apply_async 会立即返回一个 AsyncResult 对象。如果我按以下方式收集这些对象,并仅在所有工作人员完成后才使用 get(),假设值将按照调用函数的顺序是否安全? o
我正在尝试使用 multiprocessing模块,更具体地说是 Pool.apply_async()功能。 这段代码运行良好: import multiprocessing def do():
我们使用的是 django 1.10、Celery 4.1.0 我正在尝试使用 apply_async。这是任务: from celery import Celery app = Celery('my
我在 Python 3.7.3 中遇到一个问题,在处理大型计算任务时,我的多处理操作(使用队列、池和 apply_async)会死锁。 对于小型计算,这个多处理任务工作得很好。然而,当处理较大的进程时
我想调用 pool.apply_async(func) 并在结果可用时立即累积结果,而无需相互等待。 import multiprocessing import numpy as np chrName
我是一名优秀的程序员,十分优秀!