Migrate tests and add tests to retry_on param

Author: AlexVelezLl

kolibri/conftest.py

@@ -7,19 +7,6 @@
 TEMP_KOLIBRI_HOME = "./.pytest_kolibri_home"
 
 
-@pytest.fixture(scope="session")
-def django_db_setup(
-    request,
-    django_db_setup,
-):
-    def dispose_sqlalchemy():
-        from kolibri.core.tasks.main import connection
-
-        connection.dispose()
-
-    request.addfinalizer(dispose_sqlalchemy)
-
-
 @pytest.fixture(scope="session", autouse=True)
 def global_fixture():
     if not os.path.exists(TEMP_KOLIBRI_HOME):

kolibri/core/auth/test/test_auth_tasks.py

@@ -37,7 +37,6 @@
 from kolibri.core.tasks.exceptions import JobRunning
 from kolibri.core.tasks.job import Job
 from kolibri.core.tasks.job import State
-from kolibri.utils.time_utils import naive_utc_datetime
 
 
 DUMMY_PASSWORD = "password"
@@ -64,7 +63,7 @@ def fake_job(**kwargs):
 
 
 class dummy_orm_job_data(object):
-    scheduled_time = datetime.datetime(year=2023, month=1, day=1, tzinfo=None)
+    scheduled_time = datetime.datetime(year=2023, month=1, day=1)
     repeat = 5
     interval = 8600
     retry_interval = 5
@@ -701,7 +700,7 @@ def test_enqueue_soud_sync_processing__future__scheduled(
         mock_soud.get_time_to_next_attempt.return_value = datetime.timedelta(seconds=30)
         mock_job = mock_job_storage.get_orm_job.return_value
         mock_job.state = State.QUEUED
-        mock_job.scheduled_time = naive_utc_datetime(timezone.now())
+        mock_job.scheduled_time = timezone.now()
         enqueue_soud_sync_processing()
         mock_task.enqueue_in.assert_not_called()
 
@@ -714,7 +713,7 @@ def test_enqueue_soud_sync_processing__future__running(
         mock_soud.get_time_to_next_attempt.return_value = datetime.timedelta(seconds=1)
         mock_job = mock_job_storage.get_orm_job.return_value
         mock_job.state = State.RUNNING
-        mock_job.scheduled_time = naive_utc_datetime(timezone.now())
+        mock_job.scheduled_time = timezone.now()
         enqueue_soud_sync_processing()
         mock_task.enqueue_in.assert_not_called()
 
@@ -727,9 +726,7 @@ def test_enqueue_soud_sync_processing__future__reschedule(
         mock_soud.get_time_to_next_attempt.return_value = datetime.timedelta(seconds=10)
         mock_job = mock_job_storage.get_orm_job.return_value
         mock_job.state = State.QUEUED
-        mock_job.scheduled_time = naive_utc_datetime(
-            timezone.now() + datetime.timedelta(seconds=15)
-        )
+        mock_job.scheduled_time = timezone.now() + datetime.timedelta(seconds=15)
         enqueue_soud_sync_processing()
         mock_task.enqueue_in.assert_called_once_with(datetime.timedelta(seconds=10))
 
@@ -743,9 +740,7 @@ def test_enqueue_soud_sync_processing__completed__enqueue(
         mock_job = mock_job_storage.get_orm_job.return_value
         mock_job.state = State.COMPLETED
         # far in the past
-        mock_job.scheduled_time = naive_utc_datetime(
-            timezone.now() - datetime.timedelta(seconds=100)
-        )
+        mock_job.scheduled_time = timezone.now() - datetime.timedelta(seconds=100)
         enqueue_soud_sync_processing()
         mock_task.enqueue_in.assert_called_once_with(datetime.timedelta(seconds=10))
 
@@ -759,9 +754,7 @@ def test_enqueue_soud_sync_processing__race__already_running(
         mock_job = mock_job_storage.get_orm_job.return_value
         mock_job.state = State.COMPLETED
         # far in the past
-        mock_job.scheduled_time = naive_utc_datetime(
-            timezone.now() - datetime.timedelta(seconds=100)
-        )
+        mock_job.scheduled_time = timezone.now() - datetime.timedelta(seconds=100)
         mock_task.enqueue_in.side_effect = JobRunning()
         enqueue_soud_sync_processing()
         mock_task.enqueue_in.assert_called_once_with(datetime.timedelta(seconds=10))

kolibri/core/tasks/test/base.py

@@ -6,8 +6,6 @@
 from kolibri.core.tasks.exceptions import JobNotFound
 from kolibri.core.tasks.job import Job
 from kolibri.core.tasks.job import State
-from kolibri.core.tasks.storage import Storage
-from kolibri.core.tasks.test.base import connection
 from kolibri.core.tasks.utils import callable_to_import_path
 from kolibri.core.tasks.utils import get_current_job
 from kolibri.core.tasks.utils import import_path_to_callable
@@ -17,12 +15,11 @@
 
 @pytest.fixture
 def storage_fixture():
-    with connection() as conn:
-        e = Worker(connection=conn)
-        b = Storage(conn)
-        b.clear(force=True)
-        yield b
-        e.shutdown()
+    e = Worker()
+    b = e.storage
+    b.clear(force=True)
+    yield b
+    e.shutdown()
 
 
 @pytest.fixture
@@ -162,6 +159,7 @@ def update_progress_cancelable_job():
             return
 
 
+@pytest.mark.django_db(databases="__all__", transaction=True)
 class TestJobStorage(object):
     def test_does_not_enqueue_a_function(self, storage_fixture):
         try:

kolibri/core/tasks/test/taskrunner/test_job_running.py

@@ -2,27 +2,31 @@
 
 import pytest
 
+from kolibri.core.tasks.decorators import register_task
 from kolibri.core.tasks.exceptions import JobRunning
 from kolibri.core.tasks.job import Job
 from kolibri.core.tasks.storage import Storage
-from kolibri.core.tasks.test.base import connection
 from kolibri.utils.time_utils import local_now
-from kolibri.utils.time_utils import naive_utc_datetime
 
 
 @pytest.fixture
 def job_storage():
-    with connection() as c:
-        s = Storage(connection=c)
-        s.clear(force=True)
-        yield s
-        s.clear(force=True)
+    s = Storage()
+    s.clear(force=True)
+    yield s
+    s.clear(force=True)
 
 
+@register_task
+def add(x, y):
+    return x + y
+
+
+@pytest.mark.django_db(databases="__all__")
 class TestScheduler(object):
     @pytest.fixture
     def job(self):
-        return Job(id)
+        return Job(add)
 
     def test_enqueue_at_a_function(self, job_storage, job):
         job_id = job_storage.enqueue_at(local_now(), job)
@@ -34,10 +38,9 @@ def test_enqueue_at_a_function_sets_time(self, job_storage, job):
         now = local_now()
         job_id = job_storage.enqueue_at(now, job)
 
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            scheduled_time = scheduled_job.scheduled_time
-        assert scheduled_time == naive_utc_datetime(now)
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        scheduled_time = scheduled_job.scheduled_time
+        assert scheduled_time == now
 
     def test_enqueue_at_preserves_extra_metadata(self, job_storage, job):
         metadata = {"saved": True}
@@ -59,19 +62,17 @@ def test_enqueue_in_a_function_sets_time(self, job_storage, job):
         job_storage._now = lambda: now
         job_id = job_storage.enqueue_in(diff, job)
 
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            scheduled_time = scheduled_job.scheduled_time
-        assert scheduled_time == naive_utc_datetime(now) + diff
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        scheduled_time = scheduled_job.scheduled_time
+        assert scheduled_time == now + diff
 
     def test_schedule_a_function_sets_time(self, job_storage, job):
         now = local_now()
         job_id = job_storage.schedule(now, job)
 
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            scheduled_time = scheduled_job.scheduled_time
-        assert scheduled_time == naive_utc_datetime(now)
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        scheduled_time = scheduled_job.scheduled_time
+        assert scheduled_time == now
 
     def test_schedule_a_function_gives_value_error_without_datetime(
         self, job_storage, job
@@ -112,9 +113,8 @@ def test_scheduled_repeating_function_sets_endless_repeat_new_job(
         job_id = job_storage.schedule(now, job, interval=1000, repeat=None)
         job_storage.complete_job(job_id)
         job_storage.reschedule_finished_job_if_needed(job_id)
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            repeat = scheduled_job.repeat
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        repeat = scheduled_job.repeat
         assert repeat is None
 
     def test_scheduled_repeating_function_enqueues_job(self, job_storage, job):
@@ -131,9 +131,8 @@ def test_scheduled_repeating_function_sets_new_job_with_one_fewer_repeats(
         job_id = job_storage.schedule(now, job, interval=1000, repeat=1)
         job_storage.complete_job(job_id)
         job_storage.reschedule_finished_job_if_needed(job_id)
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            repeat = scheduled_job.repeat
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        repeat = scheduled_job.repeat
         assert repeat == 0
 
     def test_scheduled_repeating_function_sets_new_job_at_interval(
@@ -144,12 +143,9 @@ def test_scheduled_repeating_function_sets_new_job_at_interval(
         job_storage._now = lambda: now
         job_storage.complete_job(job_id)
         job_storage.reschedule_finished_job_if_needed(job_id)
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            scheduled_time = scheduled_job.scheduled_time
-        assert scheduled_time == naive_utc_datetime(now) + datetime.timedelta(
-            seconds=1000
-        )
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        scheduled_time = scheduled_job.scheduled_time
+        assert scheduled_time == now + datetime.timedelta(seconds=1000)
 
     def test_scheduled_repeating_function_failure_sets_new_job_at_retry_interval(
         self, job_storage, job
@@ -161,17 +157,16 @@ def test_scheduled_repeating_function_failure_sets_new_job_at_retry_interval(
         job_storage._now = lambda: now
         job_storage.mark_job_as_failed(job_id, "Exception", "Traceback")
         job_storage.reschedule_finished_job_if_needed(job_id)
-        with job_storage.session_scope() as session:
-            _, scheduled_job = job_storage._get_job_and_orm_job(job_id, session)
-            scheduled_time = scheduled_job.scheduled_time
-        assert scheduled_time == naive_utc_datetime(now) + datetime.timedelta(seconds=5)
+        _, scheduled_job = job_storage._get_job_and_orm_job(job_id)
+        scheduled_time = scheduled_job.scheduled_time
+        assert scheduled_time == now + datetime.timedelta(seconds=5)
 
 
 class TestReschedule(TestScheduler):
     @pytest.fixture
     def job(self, job_storage):
         now = local_now()
-        job_id = job_storage.schedule(now, Job(id), interval=1, repeat=123)
+        job_id = job_storage.schedule(now, Job(add), interval=1, repeat=123)
         return job_storage.get_job(job_id)
 
     def test_reschedule_a_function_gives_job_running_error(self, job_storage, job):