gva/gnuviechadmin/taskresults/tests/management/commands/test_fetch_taskresults.py

67 lines
2.3 KiB
Python
Raw Normal View History

"""
This module provides tests for the
:py:mod:`taskresults.management.commands.fetch_taskresults` Django management
command.
"""
from unittest.mock import MagicMock, patch
from django.test import TestCase
from taskresults.management.commands.fetch_taskresults import Command
from taskresults.models import TaskResult
TEST_TASK_UUID = "3120f6a8-2665-4fa3-a785-79efd28bfe92"
TEST_TASK_NAME = "test.task"
TEST_TASK_RESULT = "4ll y0ur b453 4r3 b3l0ng t0 u5"
@patch("taskresults.models.app.AsyncResult")
class FetchTaskResultsCommandTest(TestCase):
def test_handle_unfinished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertFalse(tr.finished)
self.assertEqual(tr.result, "")
self.assertEqual(tr.state, "")
aresult = asyncresult.return_value
aresult.state = "PENDING"
aresult.ready.return_value = False
2023-04-29 13:10:25 +02:00
Command().handle(verbosity=0)
tr = TaskResult.objects.get(task_id=TEST_TASK_UUID)
asyncresult.assert_called_with(TEST_TASK_UUID)
aresult.ready.assert_called_with()
self.assertFalse(tr.finished)
self.assertEqual(tr.result, "")
self.assertEqual(tr.state, "PENDING")
def test_handle_finished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertFalse(tr.finished)
self.assertEqual(tr.result, "")
self.assertEqual(tr.state, "")
aresult = asyncresult.return_value
aresult.state = "SUCCESS"
aresult.ready.return_value = True
aresult.get.return_value = TEST_TASK_RESULT
2023-04-29 13:10:25 +02:00
Command().handle(verbosity=0)
tr = TaskResult.objects.get(task_id=TEST_TASK_UUID)
asyncresult.assert_called_with(TEST_TASK_UUID)
aresult.ready.assert_called_with()
aresult.get.assert_called_with(propagate=False, timeout=5)
self.assertTrue(tr.finished)
self.assertEqual(tr.result, TEST_TASK_RESULT)
self.assertEqual(tr.state, "SUCCESS")