diff --git a/setup.cfg b/setup.cfg index 7bb61ca..5752f88 100644 --- a/setup.cfg +++ b/setup.cfg @@ -7,4 +7,4 @@ universal = 1 [flake8] max-line-length=120 -ignore=E731 \ No newline at end of file +ignore=E731 diff --git a/tests/fixtures.py b/tests/fixtures.py index 9173989..f9d1eb1 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -146,3 +146,7 @@ def run_dummy_heroku_worker(sandbox, _imminent_shutdown_delay): w = TestHerokuWorker(Queue('dummy')) w.main_work_horse(None, None) + + +class DummyQueue(object): + pass diff --git a/tests/test_connection.py b/tests/test_connection.py index 505f129..c04358b 100644 --- a/tests/test_connection.py +++ b/tests/test_connection.py @@ -2,7 +2,8 @@ from __future__ import (absolute_import, division, print_function, unicode_literals) -from rq import Connection, Queue +from rq import Connection, Queue, use_connection, get_current_connection, pop_connection +from rq.connections import NoRedisConnectionException from tests import find_empty_redis_database, RQTestCase from tests.fixtures import do_nothing @@ -38,3 +39,30 @@ class TestConnectionInheritance(RQTestCase): job2 = q2.enqueue(do_nothing) self.assertEqual(q1.connection, job1.connection) self.assertEqual(q2.connection, job2.connection) + + +class TestConnectionHelpers(RQTestCase): + def test_use_connection(self): + """Test function use_connection works as expected.""" + conn = new_connection() + use_connection(conn) + + self.assertEqual(conn, get_current_connection()) + + use_connection() + + self.assertNotEqual(conn, get_current_connection()) + + use_connection(self.testconn) # Restore RQTestCase connection + + with self.assertRaises(AssertionError): + with Connection(new_connection()): + use_connection() + with Connection(new_connection()): + use_connection() + + def test_resolve_connection_raises_on_no_connection(self): + """Test function resolve_connection raises if there is no connection.""" + pop_connection() + with self.assertRaises(NoRedisConnectionException): + Queue() diff --git a/tests/test_utils.py b/tests/test_utils.py index d3d9ef9..529c61e 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,9 +1,11 @@ # -*- coding: utf-8 -*- from __future__ import (absolute_import, division, print_function, unicode_literals) - -from tests import RQTestCase -from rq.utils import parse_timeout +import re +import datetime +from tests import RQTestCase, fixtures +from rq.utils import parse_timeout, first, is_nonstring_iterable, ensure_list, utcparse, backend_class +from rq.exceptions import TimeoutFormatError class TestUtils(RQTestCase): @@ -15,3 +17,50 @@ class TestUtils(RQTestCase): self.assertEqual(720, parse_timeout('12m')) self.assertEqual(3600, parse_timeout('1h')) self.assertEqual(3600, parse_timeout('1H')) + + def test_parse_timeout_coverage_scenarios(self): + """Test parse_timeout edge cases for coverage""" + timeouts = ['h12', 'h', 'm', 's', '10k'] + + self.assertEqual(None, parse_timeout(None)) + with self.assertRaises(TimeoutFormatError): + for timeout in timeouts: + parse_timeout(timeout) + + def test_first(self): + """Ensure function first works correctly""" + self.assertEqual(42, first([0, False, None, [], (), 42])) + self.assertEqual(None, first([0, False, None, [], ()])) + self.assertEqual('ohai', first([0, False, None, [], ()], default='ohai')) + self.assertEqual('bc', first(re.match(regex, 'abc') for regex in ['b.*', 'a(.*)']).group(1)) + self.assertEqual(4, first([1, 1, 3, 4, 5], key=lambda x: x % 2 == 0)) + + def test_is_nonstring_iterable(self): + """Ensure function is_nonstring_iterable works correctly""" + self.assertEqual(True, is_nonstring_iterable([])) + self.assertEqual(False, is_nonstring_iterable('test')) + self.assertEqual(True, is_nonstring_iterable({})) + self.assertEqual(True, is_nonstring_iterable(())) + + def test_ensure_list(self): + """Ensure function ensure_list works correctly""" + self.assertEqual([], ensure_list([])) + self.assertEqual(['test'], ensure_list('test')) + self.assertEqual({}, ensure_list({})) + self.assertEqual((), ensure_list(())) + + def test_utcparse(self): + """Ensure function utcparse works correctly""" + utc_formated_time = '2017-08-31T10:14:02Z' + utc_compat_formated_time = '2017-08-31T10:20:56.226733+00:00' + + self.assertEqual(datetime.datetime(2017, 8, 31, 10, 14, 2), utcparse(utc_formated_time)) + self.assertEqual(datetime.datetime(2017, 8, 31, 10, 20, 56, 226733), utcparse(utc_compat_formated_time)) + + def test_backend_class(self): + """Ensure function backend_class works correctly""" + self.assertEqual(fixtures.DummyQueue, backend_class(fixtures, 'DummyQueue')) + self.assertNotEqual(fixtures.say_pid, backend_class(fixtures, 'DummyQueue')) + self.assertEqual(fixtures.DummyQueue, backend_class(fixtures, 'DummyQueue', override=fixtures.DummyQueue)) + self.assertEqual(fixtures.DummyQueue, + backend_class(fixtures, 'DummyQueue', override='tests.fixtures.DummyQueue'))