gva/gnuviechadmin/taskresults/tests/test_models.py

84 lines
3 KiB
Python
Raw Normal View History

"""
This module provides tests for :py:mod:`taskresults.models`.
"""
from unittest.mock import MagicMock, patch
2015-02-01 03:00:14 +01:00
from django.test import TestCase
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"
2015-02-01 03:00:14 +01:00
class TaskResultTest(TestCase):
@patch("taskresults.models.app.AsyncResult")
def test_update_taskstatus_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)
2015-02-01 03:00:14 +01:00
self.assertFalse(tr.finished)
mymock = asyncresult.return_value
mymock.state = "PENDING"
mymock.ready.return_value = False
2015-02-01 03:00:14 +01:00
tr.fetch_result()
mymock.get.assert_not_called()
self.assertFalse(tr.finished)
2015-02-01 03:00:14 +01:00
@patch("taskresults.models.app.AsyncResult")
def test_update_taskstatus_finished(self, asyncresult):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
sigmock = MagicMock()
sigmock.apply_async.return_value = resultmock
aresult = asyncresult.return_value
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, sigmock)
self.assertFalse(tr.finished)
aresult = asyncresult.return_value
aresult.state = "SUCCESS"
aresult.ready.return_value = True
aresult.get.return_value = TEST_TASK_RESULT
2015-02-01 03:00:14 +01:00
tr.fetch_result()
self.assertTrue(aresult.get.called_with())
self.assertEqual(aresult.get.call_count, 1)
2015-02-01 03:00:14 +01:00
self.assertTrue(tr.finished)
self.assertEqual(tr.result, str(TEST_TASK_RESULT))
2015-02-01 03:00:14 +01:00
tr.fetch_result()
self.assertEqual(aresult.get.call_count, 1)
2015-02-01 03:00:14 +01:00
self.assertTrue(tr.finished)
self.assertEqual(tr.result, str(TEST_TASK_RESULT))
def test___str__(self):
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.assertEqual(
str(tr),
"{name} ({taskid}): no".format(name=TEST_TASK_NAME, taskid=TEST_TASK_UUID),
)
tr.finished = True
self.assertEqual(
str(tr),
"{name} ({taskid}): yes".format(name=TEST_TASK_NAME, taskid=TEST_TASK_UUID),
)
2015-02-01 03:00:14 +01:00
TEST_RESULT = MagicMock()
TEST_RESULT.task_id = TEST_TASK_UUID
TEST_RESULT.task_name = TEST_TASK_NAME
TEST_RESULT.ready.return_value = False
class TaskResultManagerTest(TestCase):
def test_create_task_result(self):
resultmock = MagicMock(task_id=TEST_TASK_UUID)
mock = MagicMock()
mock.apply_async.return_value = resultmock
tr = TaskResult.objects.create_task_result(TEST_TASK_NAME, mock)
2015-02-01 03:00:14 +01:00
self.assertIsInstance(tr, TaskResult)
self.assertEqual(tr.task_id, TEST_TASK_UUID)
self.assertEqual(tr.creator, TEST_TASK_NAME)