blocks|key|747154|text|在队列中仍有作业时，您需要测试暂停。为此，您可以检查Queue.is_empty()，如果队列中仍然有作业，则可以挂起执行：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|747155|import+time
from+django.utils.unittest+import+TestCase
import+django_rq

class+TestQueue(TestCase):

def+test_something(self):
++++#+simulate+some+User+actions+which+will+queue+up+some+tasks

++++#+Wait+for+the+queued+tasks+to+run
++++queue+=+django_rq.get_queue('default')
++++while+not+queue.is_empty():
++++++++time.sleep(5)+#+adjust+this+depending+on+how+long+your+tasks+take+to+execute

++++#+queued+tasks+are+done,+check+state+of+the+DB
++++self.assert(.....)|code-block|syntax|javascript|747156|entityMap^0|Q|G|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@$9|N|A|O|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|P|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|Q|8|@]|D|@]|E|$]]]|L|$]]

You'll need your tests to pause while there are still jobs in the queue. To do this, you can check <code>Queue.is_empty()</code>, and suspend execution if there are still jobs in the queue:

<pre><code>import time
from django.utils.unittest import TestCase
import django_rq

class TestQueue(TestCase):

def test_something(self):
 # simulate some User actions which will queue up some tasks

 # Wait for the queued tasks to run
 queue = django_rq.get_queue('default')
 while not queue.is_empty():
 time.sleep(5) # adjust this depending on how long your tasks take to execute

 # queued tasks are done, check state of the DB
 self.assert(.....)
</code></pre>

blocks|key|1508963|text|我刚刚找到了django-rq，它允许您在一个测试环境中旋转一个工作人员，该环境执行队列中的任何任务，然后退出。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1508964|from+django.test+impor+TestCase
from+django_rq+import+get_worker

class+MyTest(TestCase):
++++def+test_something_that_creates_jobs(self):
++++++++...++++++++++++++++++++++#+Stuff+that+init+jobs.
++++++++get_worker().work(burst=True)++#+Processes+all+jobs+then+stop.
++++++++...++++++++++++++++++++++#+Asserts+that+the+job+stuff+is+done.|code-block|syntax|javascript|1508965|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/ui/django-rq^0|6|9|6|9|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@$9|T|A|U|B|C]]|D|@$9|V|A|W|1|X]]|E|$]]|$1|F|3|G|5|H|7|Y|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|Z|8|@]|D|@]|E|$]]]|L|$M|$5|N|O|P|E|$Q|R]]]]

I just found <a href="https://github.com/ui/django-rq" rel="noreferrer"><code>django-rq</code></a>, which allows you to spin up a worker in a test environment that executes any tasks on the queue and then quits.

<pre><code>from django.test impor TestCase
from django_rq import get_worker

class MyTest(TestCase):
 def test_something_that_creates_jobs(self):
 ... # Stuff that init jobs.
 get_worker().work(burst=True) # Processes all jobs then stop.
 ... # Asserts that the job stuff is done.
</code></pre>

blocks|key|1029603|text|我把我的rq测试分成几个部分。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1029604|测试我是否正确地将东西添加到队列中(使用模拟)。|ordered-list-item|1029605|假设如果将某些内容添加到队列中，则最终将对其进行处理。(rq的测试套件应该涵盖这一点)。|1029606|测试，给定正确的输入，我的任务就像预期的那样工作。(正常代码测试)。|1029607|正在测试的代码：|1029608|def+handle(self,+*args,+**options):
++++uid+=+options.get('user_id')

++++#+@@@+Need+to+exclude+out+users+who+have+gotten+an+email+within+$window
++++#+days.
++++if+uid+is+None:
++++++++uids+=+User.objects.filter(is_active=True,+userprofile__waitlisted=False).values_list('id',+flat=True)
++++else:
++++++++uids+=+[uid]

++++q+=+rq.Queue(connection=redis.Redis())

++++for+user_id+in+uids:
++++++++q.enqueue(mail_user,+user_id)|code-block|syntax|javascript|1029609|我的测试：|1029610|class+DjangoMailUsersTest(DjangoTestCase):
++++def+setUp(self):
++++++++self.cmd+=+MailUserCommand()

++++@patch('redis.Redis')
++++@patch('rq.Queue')
++++def+test_no_userid_queues_all_userids(self,+queue,+_):
++++++++u1+=+UserF.create(userprofile__waitlisted=False)
++++++++u2+=+UserF.create(userprofile__waitlisted=False)
++++++++self.cmd.handle()
++++++++self.assertItemsEqual(queue.return_value.enqueue.mock_calls,
++++++++++++++++++++++++++++++[call(ANY,+u1.pk),+call(ANY,+u2.pk)])

++++@patch('redis.Redis')
++++@patch('rq.Queue')
++++def+test_waitlisted_people_excluded(self,+queue,+_):
++++++++u1+=+UserF.create(userprofile__waitlisted=False)
++++++++UserF.create(userprofile__waitlisted=True)
++++++++self.cmd.handle()
++++++++self.assertItemsEqual(queue.return_value.enqueue.mock_calls,+[call(ANY,+u1.pk)])|1029611|entityMap^0|4|2|0|0|S|2|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|Z|8|@$9|10|A|11|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|12|8|@]|D|@]|E|$]]|$1|I|3|J|5|H|7|13|8|@$9|14|A|15|B|C]]|D|@]|E|$]]|$1|K|3|L|5|H|7|16|8|@]|D|@]|E|$]]|$1|M|3|N|5|6|7|17|8|@]|D|@]|E|$]]|$1|O|3|P|5|Q|7|18|8|@]|D|@]|E|$R|S]]|$1|T|3|U|5|6|7|19|8|@]|D|@]|E|$]]|$1|V|3|W|5|Q|7|1A|8|@]|D|@]|E|$R|S]]|$1|X|3|-4|5|6|7|1B|8|@]|D|@]|E|$]]]|Y|$]]

I separated my <code>rq</code> tests into a few pieces.

<ol>
<li>Test that I'm correctly adding things to the queue (using mocks).</li>
<li>Assume that if something gets added to the queue, it will eventually be processed. (<code>rq</code>'s test suite should cover this).</li>
<li>Test, given the correct input, my tasks work as expected. (normal code tests).</li>
</ol>


Code being tested:

<pre><code>def handle(self, *args, **options):
 uid = options.get('user_id')

 # @@@ Need to exclude out users who have gotten an email within $window
 # days.
 if uid is None:
 uids = User.objects.filter(is_active=True, userprofile__waitlisted=False).values_list('id', flat=True)
 else:
 uids = [uid]

 q = rq.Queue(connection=redis.Redis())

 for user_id in uids:
 q.enqueue(mail_user, user_id)
</code></pre>

My tests:

<pre><code>class DjangoMailUsersTest(DjangoTestCase):
 def setUp(self):
 self.cmd = MailUserCommand()

 @patch('redis.Redis')
 @patch('rq.Queue')
 def test_no_userid_queues_all_userids(self, queue, _):
 u1 = UserF.create(userprofile__waitlisted=False)
 u2 = UserF.create(userprofile__waitlisted=False)
 self.cmd.handle()
 self.assertItemsEqual(queue.return_value.enqueue.mock_calls,
 [call(ANY, u1.pk), call(ANY, u2.pk)])

 @patch('redis.Redis')
 @patch('rq.Queue')
 def test_waitlisted_people_excluded(self, queue, _):
 u1 = UserF.create(userprofile__waitlisted=False)
 UserF.create(userprofile__waitlisted=True)
 self.cmd.handle()
 self.assertItemsEqual(queue.return_value.enqueue.mock_calls, [call(ANY, u1.pk)])
</code></pre>

blocks|key|1029623|text|我学习了一块地，它让你做到了：|type|unstyled|depth|inlineStyleRanges|entityRanges|offset|length|data|1029624|from+django.test+impor+TestCase
from+django_rq+import+get_queue

class+MyTest(TestCase):
++++def+test_something_that_creates_jobs(self):
++++++++queue+=+get_queue(async=False)
++++++++queue.enqueue(func)+#+func+will+be+executed+right+away
++++++++#+Test+for+job+completion|code-block|syntax|javascript|1029625|这将使测试RQ作业更容易。希望这能帮上忙！|1029626|entityMap|0|LINK|mutability|MUTABLE|url|https://github.com/ui/django-rq/commit/a57ee600a18a9c7cc796efc0cbe2d8a1eab4909c^0|4|3|0|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@]|9|@$A|T|B|U|1|V]]|C|$]]|$1|D|3|E|5|F|7|W|8|@]|9|@]|C|$G|H]]|$1|I|3|J|5|6|7|X|8|@]|9|@]|C|$]]|$1|K|3|-4|5|6|7|Y|8|@]|9|@]|C|$]]]|L|$M|$5|N|O|P|C|$Q|R]]]]

I commited <a href="https://github.com/ui/django-rq/commit/a57ee600a18a9c7cc796efc0cbe2d8a1eab4909c" rel="nofollow">a patch</a> that lets you do:

<pre><code>from django.test impor TestCase
from django_rq import get_queue

class MyTest(TestCase):
 def test_something_that_creates_jobs(self):
 queue = get_queue(async=False)
 queue.enqueue(func) # func will be executed right away
 # Test for job completion
</code></pre>

This should make testing RQ jobs easier. Hope that helps!

blocks|key|1029648|text|以防万一这对任何人都有帮助。我使用一个带有自定义模拟对象的修补程序来执行将立即运行的队列。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1029649|#patch+django_rq.get_queue
with+patch('django_rq.get_queue',+return_value=MockBulkJobGetQueue())+as+mock_django_rq_get_queue:
++++#Perform+web+operation+that+starts+job.+In+my+case+a+post+to+a+url|code-block|syntax|javascript|1029650|然后，模拟对象只有一个方法：|1029651|class+MockBulkJobGetQueue(object):

++++def+enqueue(self,+f,+*args,+**kwargs):
++++++++#+Call+the+function
++++++++f(
++++++++++++**kwargs.pop('kwargs',+None)
++++++++)|1029652|entityMap^0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|M|8|@]|9|@]|A|$]]|$1|B|3|C|5|D|7|N|8|@]|9|@]|A|$E|F]]|$1|G|3|H|5|6|7|O|8|@]|9|@]|A|$]]|$1|I|3|J|5|D|7|P|8|@]|9|@]|A|$E|F]]|$1|K|3|-4|5|6|7|Q|8|@]|9|@]|A|$]]]|L|$]]

Just in case this would be helpful to anyone. I used a patch with a custom mock object to do the enqueue that would run right away

<pre><code>#patch django_rq.get_queue
with patch('django_rq.get_queue', return_value=MockBulkJobGetQueue()) as mock_django_rq_get_queue:
 #Perform web operation that starts job. In my case a post to a url
</code></pre>

Then the mock object just had one method:

<pre><code>class MockBulkJobGetQueue(object):

 def enqueue(self, f, *args, **kwargs):
 # Call the function
 f(
 **kwargs.pop('kwargs', None)
 )
</code></pre>

blocks|key|1029685|text|我在这个案例中所做的就是检测我是否在测试，并在测试期间使用假货。最后，在测试本身中，我以同步模式将redis工作者任务排队：|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1029686|首先，定义一个检测是否正在测试的函数：|1029687|TESTING+=+len(sys.argv)+>+1+and+sys.argv[1]+==+'test'

def+am_testing():
++++return+TESTING|code-block|syntax|javascript|1029688|然后，在使用redis对任务排队的文件中，以这种方式管理队列。如果需要，可以扩展get_queue以指定队列名称：|1029689|if+am_testing():
++++from+fakeredis+import+FakeStrictRedis+
++++from+rq+import+Queue
++++def+get_queue():
++++++++return+Queue(connection=FakeStrictRedis())

else:
++++import+django_rq
++++def+get_queue():
++++++++return+django_rq.get_queue()|1029690|然后，将您的任务排成这样的队列：|1029691|queue+=+get_queue()
queue.enqueue(task_mytask,+arg1,+arg2)|1029692|最后，在您的测试程序中，以同步模式运行您正在测试的任务，以便它在与您的测试相同的进程中运行。实际上，我先清理假货队列，但我认为没有必要，因为没有工人：|1029693|from+rq+import+Queue
from+fakeredis+import+FakeStrictRedis

FakeStrictRedis().flushall()
queue+=+Queue(async=False,+connection=FakeStrictRedis())
queue.enqueue(task_mytask,+arg1,+arg2)|1029694|我的settings.py具有正常的django_redis设置，因此django_rq.getqueue()在部署时使用这些设置：|1029695|RQ_QUEUES+=+{
++++'default':+{
++++++++'HOST':+env_var('REDIS_HOST'),
++++++++'PORT':+6379,
++++++++'DB':+0,
++++++++#+'PASSWORD':+'some-password',
++++++++'DEFAULT_TIMEOUT':+360,
++++},
++++'high':+{
++++++++'HOST':+env_var('REDIS_HOST'),
++++++++'PORT':+6379,
++++++++'DB':+0,
++++++++'DEFAULT_TIMEOUT':+500,
++++},
++++'low':+{
++++++++'HOST':+env_var('REDIS_HOST'),
++++++++'PORT':+6379,
++++++++'DB':+0,
++++}
}|1029696|entityMap^0|0|0|0|0|0|0|0|0|0|0|0^^$0|@$1|2|3|4|5|6|7|10|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|11|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|12|8|@]|9|@]|A|$G|H]]|$1|I|3|J|5|6|7|13|8|@]|9|@]|A|$]]|$1|K|3|L|5|F|7|14|8|@]|9|@]|A|$G|H]]|$1|M|3|N|5|6|7|15|8|@]|9|@]|A|$]]|$1|O|3|P|5|F|7|16|8|@]|9|@]|A|$G|H]]|$1|Q|3|R|5|6|7|17|8|@]|9|@]|A|$]]|$1|S|3|T|5|F|7|18|8|@]|9|@]|A|$G|H]]|$1|U|3|V|5|6|7|19|8|@]|9|@]|A|$]]|$1|W|3|X|5|F|7|1A|8|@]|9|@]|A|$G|H]]|$1|Y|3|-4|5|6|7|1B|8|@]|9|@]|A|$]]]|Z|$]]

what I've done for this case is to detect if I'm testing, and use fakeredis during tests. finally, in the test itself, I enqueue the redis worker task in synch mode:

first, define a function that detects if you're testing:

<pre><code>TESTING = len(sys.argv) &gt; 1 and sys.argv[1] == 'test'

def am_testing():
 return TESTING
</code></pre>

then in your file that uses redis to queue up tasks, manage the queue this way. 
 you could extend get_queue to specify a queue name if needed:

<pre><code>if am_testing():
 from fakeredis import FakeStrictRedis 
 from rq import Queue
 def get_queue():
 return Queue(connection=FakeStrictRedis())

else:
 import django_rq
 def get_queue():
 return django_rq.get_queue()
</code></pre>

then, enqueue your task like so:

<pre><code>queue = get_queue()
queue.enqueue(task_mytask, arg1, arg2)
</code></pre>

finally, in your test program, run the task you are testing in synch mode, so that it runs in the same process as your test. As a matter of practice, I first clear the fakeredis queue, but I don't think its necessary since there are no workers:

<pre><code>from rq import Queue
from fakeredis import FakeStrictRedis

FakeStrictRedis().flushall()
queue = Queue(async=False, connection=FakeStrictRedis())
queue.enqueue(task_mytask, arg1, arg2)
</code></pre>

my settings.py has the normal django_redis settings, so django_rq.getqueue() uses these when deployed:

<pre><code>RQ_QUEUES = {
 'default': {
 'HOST': env_var('REDIS_HOST'),
 'PORT': 6379,
 'DB': 0,
 # 'PASSWORD': 'some-password',
 'DEFAULT_TIMEOUT': 360,
 },
 'high': {
 'HOST': env_var('REDIS_HOST'),
 'PORT': 6379,
 'DB': 0,
 'DEFAULT_TIMEOUT': 500,
 },
 'low': {
 'HOST': env_var('REDIS_HOST'),
 'PORT': 6379,
 'DB': 0,
 }
}
</code></pre>

blocks|key|1029718|text|上面的答案都没有真正解决如何在没有安装redis和使用django设置的情况下进行测试。我发现，在测试中包含以下代码并不会影响项目本身，但会给出所需的一切。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1029719|代码使用fakeredis来假装有一个Redis服务可用，在reads读取设置之前设置连接。连接必须相同，因为在伪造连接中，连接不共享状态。因此，重用它是一个单例对象。|1029720|from+fakeredis+import+FakeStrictRedis,+FakeRedis

class+FakeRedisConn:
++++"""Singleton+FakeRedis+connection."""

++++def+__init__(self):
++++++++self.conn+=+None

++++def+__call__(self,+_,+strict):
++++++++if+not+self.conn:
++++++++++++self.conn+=+FakeStrictRedis()+if+strict+else+FakeRedis()
++++++++return+self.conn


django_rq.queues.get_redis_connection+=+FakeRedisConn()

def+test_case():
+++...|code-block|syntax|javascript|1029721|entityMap^0|0|0|0^^$0|@$1|2|3|4|5|6|7|K|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|L|8|@]|9|@]|A|$]]|$1|D|3|E|5|F|7|M|8|@]|9|@]|A|$G|H]]|$1|I|3|-4|5|6|7|N|8|@]|9|@]|A|$]]]|J|$]]

None of the answers above really solved how to test without having redis installed and using django settings. I found including the following code in the tests does not impact the project itself yet gives everything needed.
The code uses fakeredis to pretend there is a Redis service available, set up the connection before RQ Django reads the settings.
The connection must be the same because in fakeredis connections
do not share the state. Therefore, it is a singleton object to reuse it.
<pre class="lang-py prettyprint-override"><code>from fakeredis import FakeStrictRedis, FakeRedis

class FakeRedisConn:
 &quot;&quot;&quot;Singleton FakeRedis connection.&quot;&quot;&quot;

 def __init__(self):
 self.conn = None

 def __call__(self, _, strict):
 if not self.conn:
 self.conn = FakeStrictRedis() if strict else FakeRedis()
 return self.conn


django_rq.queues.get_redis_connection = FakeRedisConn()

def test_case():
 ...

</code></pre>

I'll start using django-rq in my project.

<blockquote>
 Django integration with RQ, a Redis based Python queuing library.
</blockquote>

What is the best practice of testing django apps which is using RQ?

For example, if I want to test my app as a black box, after User makes some actions I want to execute all jobs in current Queue, and then check all results in my DB. How can I do it in my django-tests?

Best practice of testing django-rq ( python-rq ) in Django

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋 

腾讯云代码助手

CODING DevOps

Cloud Studio

SDK中心

API中心

命令行工具

我将在我的项目中开始使用django-rq。Django与基于Redis的Python队列库RQ集成。使用RQ测试django应用程序的最佳实践是什么？例如，如果我想将我的应用程序作为一个黑匣子进行测试，在用户执行一些操作之后，我想要执行当前队列中的所有作业，然后检查我的DB中的所有结果。我怎么能在我的django测试...

问在Django中测试django-rq (Python)的最佳实践
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问在Django中测试django-rq (Python)的最佳实践EN