rq/tests/test_fixtures.py

from rq import Queue

from tests import RQTestCase, fixtures


class TestFixtures(RQTestCase):
    def test_rpush_fixture(self):
        fixtures.rpush('foo', 'bar')
        assert self.testconn.lrange('foo', 0, 0)[0].decode() == 'bar'

    def test_start_worker_fixture(self):
        queue = Queue(name='testing', connection=self.testconn)
        queue.enqueue(fixtures.say_hello)
        conn_kwargs = self.testconn.connection_pool.connection_kwargs
        fixtures.start_worker(queue.name, conn_kwargs, 'w1', True)
        assert not queue.jobs
Multidependencies (#1397) * Also accept lists and tuples as value of `depends_on`. * The elements of the lists/tuples may be either Jobs or Job IDs. * A single Job / Job ID is still accepted as well. * Represent _all_ job dependencies in `Job.to_dict()`. We now represent the entire list, instead of just the first element. * Fix some doctext regarding plurality of dependencies. * Add unit tests for job dependencies. * One unit test establishes a pattern for checking execution order as affected by dependencies. * Another unit test applies this pattern to the new capability to name multiple dependencies. * Add unit test for new `depends_on` input formats. Also test that these are properly persisted. * Repair `Job.restore()`. Need to convert bytes back to strings when reloading `dependency_ids`. * Maintain backwards compat. in `Job.to_dict()`. Keep the old `dependency_id` (singular) key. * Provide coverage for new test fixture. * Simplify some code. Cut some superfluous `as_text()` calls left over from an earlier commit. * Check for `dependency_id` in `Job.restore()` for backwd. compat. Also eliminate use of `as_text()` here, in favor of `.decode()`. * Switch to snake case instead of camel case. * Eliminate some usages of `as_text()`. Also cut some `print` statements. * Cleanup. * Accept arbitrary iterables for `Job`'s `depends_on` kwarg. Instead of requiring a list or tuple, we now make use of `ensure_list()`. * Add test fixtures. * Provide a system to get two workers working simultaneously, using `multiprocessing`. * Define a simple job that just says whether its dependencies are met. * In `rpush`, make an option to record the name of the worker. * Improve unit tests on execution order with dependencies. These now actually have two workers going, which makes a more thorough test. * Add unit test examining `Job.dependencies_are_met()` at execution time. * Redesign dependency execution order unit tests. * Simplify assertions. * Improve doctext and formatting. * Move fixture tests to new, dedicated module `test_fixtures.py`. * Use `enqueue` instead of `enqueue_call` in new unit tests. 4 years ago			`from rq import Queue`

			`from tests import RQTestCase, fixtures`


			`class TestFixtures(RQTestCase):`
			`def test_rpush_fixture(self):`
			`fixtures.rpush('foo', 'bar')`
			`assert self.testconn.lrange('foo', 0, 0)[0].decode() == 'bar'`

			`def test_start_worker_fixture(self):`
			`queue = Queue(name='testing', connection=self.testconn)`
			`queue.enqueue(fixtures.say_hello)`
			`conn_kwargs = self.testconn.connection_pool.connection_kwargs`
			`fixtures.start_worker(queue.name, conn_kwargs, 'w1', True)`
			`assert not queue.jobs`