Jan Dittberner
e96aac82fc
This commit adds tests for the taskresults management command fetch_taskresults and completes the test coverage for taskresults.models.
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
"""
|
|
This module provides tests for :py:mod:`taskresults.models`.
|
|
|
|
"""
|
|
from __future__ import absolute_import, unicode_literals
|
|
|
|
try:
|
|
from unittest.mock import patch, MagicMock
|
|
except ImportError:
|
|
from mock import patch, MagicMock
|
|
|
|
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'
|
|
|
|
|
|
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)
|
|
self.assertFalse(tr.finished)
|
|
mymock = asyncresult.return_value
|
|
mymock.state = 'PENDING'
|
|
mymock.ready.return_value = False
|
|
tr.fetch_result()
|
|
mymock.get.assert_not_called()
|
|
self.assertFalse(tr.finished)
|
|
|
|
@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
|
|
tr.fetch_result()
|
|
self.assertTrue(aresult.get.called_with())
|
|
self.assertEqual(aresult.get.call_count, 1)
|
|
self.assertTrue(tr.finished)
|
|
self.assertEqual(tr.result, str(TEST_TASK_RESULT))
|
|
tr.fetch_result()
|
|
self.assertEqual(aresult.get.call_count, 1)
|
|
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))
|
|
|
|
|
|
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)
|
|
self.assertIsInstance(tr, TaskResult)
|
|
self.assertEqual(tr.task_id, TEST_TASK_UUID)
|
|
self.assertEqual(tr.creator, TEST_TASK_NAME)
|